add scanned metrics by date

Author: nikhilsinhaparseable

src/metadata.rs

@@ -62,10 +62,10 @@ pub fn update_stats(
         .add(size as i64);
     TOTAL_EVENTS_INGESTED_DATE
         .with_label_values(&[origin, &parsed_date])
-        .inc_by(num_rows as u64);
+        .add(num_rows as i64);
     TOTAL_EVENTS_INGESTED_SIZE_DATE
         .with_label_values(&[origin, &parsed_date])
-        .inc_by(size);
+        .add(size as i64);
 }
 
 /// In order to support backward compatability with streams created before v1.6.4,

src/metrics/mod.rs

@@ -160,8 +160,8 @@ pub static EVENTS_STORAGE_SIZE_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
-pub static TOTAL_EVENTS_INGESTED_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
-    IntCounterVec::new(
+pub static TOTAL_EVENTS_INGESTED_DATE: Lazy<IntGaugeVec> = Lazy::new(|| {
+    IntGaugeVec::new(
         Opts::new(
             "total_events_ingested_date",
             "total events ingested on a particular date",
@@ -172,8 +172,8 @@ pub static TOTAL_EVENTS_INGESTED_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
-pub static TOTAL_EVENTS_INGESTED_SIZE_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
-    IntCounterVec::new(
+pub static TOTAL_EVENTS_INGESTED_SIZE_DATE: Lazy<IntGaugeVec> = Lazy::new(|| {
+    IntGaugeVec::new(
         Opts::new(
             "total_events_ingested_size_date",
             "Total events ingested size in bytes on a particular date",
@@ -184,8 +184,8 @@ pub static TOTAL_EVENTS_INGESTED_SIZE_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
-pub static TOTAL_EVENTS_STORAGE_SIZE_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
-    IntCounterVec::new(
+pub static TOTAL_EVENTS_STORAGE_SIZE_DATE: Lazy<IntGaugeVec> = Lazy::new(|| {
+    IntGaugeVec::new(
         Opts::new(
             "total_events_storage_size_date",
             "Total events storage size in bytes on a particular date",

src/metrics/storage.rs

@@ -44,12 +44,24 @@ pub static STORAGE_FILES_SCANNED: Lazy<CounterVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
+pub static STORAGE_FILES_SCANNED_DATE: Lazy<CounterVec> = Lazy::new(|| {
+    CounterVec::new(
+        Opts::new(
+            "storage_files_scanned_date_total",
+            "Total number of files scanned in storage operations by date",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["provider", "operation", "date"],
+    )
+    .expect("metric can be created")
+});
+
 pub trait StorageMetrics {
     fn register_metrics(&self, handler: &PrometheusMetrics);
 }
 
 pub mod localfs {
-    use crate::storage::FSConfig;
+    use crate::{metrics::storage::STORAGE_FILES_SCANNED_DATE, storage::FSConfig};
 
     use super::{STORAGE_FILES_SCANNED, STORAGE_REQUEST_RESPONSE_TIME, StorageMetrics};
 
@@ -63,12 +75,16 @@ pub mod localfs {
                 .registry
                 .register(Box::new(STORAGE_FILES_SCANNED.clone()))
                 .expect("metric can be registered");
+            handler
+                .registry
+                .register(Box::new(STORAGE_FILES_SCANNED_DATE.clone()))
+                .expect("metric can be registered");
         }
     }
 }
 
 pub mod s3 {
-    use crate::storage::S3Config;
+    use crate::{metrics::storage::STORAGE_FILES_SCANNED_DATE, storage::S3Config};
 
     use super::{STORAGE_FILES_SCANNED, STORAGE_REQUEST_RESPONSE_TIME, StorageMetrics};
 
@@ -82,12 +98,16 @@ pub mod s3 {
                 .registry
                 .register(Box::new(STORAGE_FILES_SCANNED.clone()))
                 .expect("metric can be registered");
+            handler
+                .registry
+                .register(Box::new(STORAGE_FILES_SCANNED_DATE.clone()))
+                .expect("metric can be registered");
         }
     }
 }
 
 pub mod azureblob {
-    use crate::storage::AzureBlobConfig;
+    use crate::{metrics::storage::STORAGE_FILES_SCANNED_DATE, storage::AzureBlobConfig};
 
     use super::{STORAGE_FILES_SCANNED, STORAGE_REQUEST_RESPONSE_TIME, StorageMetrics};
 
@@ -101,12 +121,16 @@ pub mod azureblob {
                 .registry
                 .register(Box::new(STORAGE_FILES_SCANNED.clone()))
                 .expect("metric can be registered");
+            handler
+                .registry
+                .register(Box::new(STORAGE_FILES_SCANNED_DATE.clone()))
+                .expect("metric can be registered");
         }
     }
 }
 
 pub mod gcs {
-    use crate::storage::GcsConfig;
+    use crate::{metrics::storage::STORAGE_FILES_SCANNED_DATE, storage::GcsConfig};
 
     use super::{STORAGE_FILES_SCANNED, STORAGE_REQUEST_RESPONSE_TIME, StorageMetrics};
 
@@ -120,6 +144,10 @@ pub mod gcs {
                 .registry
                 .register(Box::new(STORAGE_FILES_SCANNED.clone()))
                 .expect("metric can be registered");
+            handler
+                .registry
+                .register(Box::new(STORAGE_FILES_SCANNED_DATE.clone()))
+                .expect("metric can be registered");
         }
     }
 }

src/query/stream_schema_provider.rs

@@ -57,7 +57,10 @@ use crate::{
     },
     event::DEFAULT_TIMESTAMP_KEY,
     hottier::HotTierManager,
-    metrics::{QUERY_CACHE_HIT, storage::STORAGE_FILES_SCANNED},
+    metrics::{
+        QUERY_CACHE_HIT,
+        storage::{STORAGE_FILES_SCANNED, STORAGE_FILES_SCANNED_DATE},
+    },
     option::Mode,
     parseable::{PARSEABLE, STREAM_EXISTS},
     storage::{ObjectStorage, ObjectStorageError, ObjectStoreFormat, STREAM_ROOT_DIRECTORY},
@@ -570,6 +573,13 @@ impl TableProvider for StandardTableProvider {
         STORAGE_FILES_SCANNED
             .with_label_values(&[PARSEABLE.storage().name(), "GET"])
             .inc_by(parquet_files_to_scan as f64);
+        STORAGE_FILES_SCANNED_DATE
+            .with_label_values(&[
+                PARSEABLE.storage().name(),
+                "GET",
+                &Utc::now().date_naive().to_string(),
+            ])
+            .inc_by(parquet_files_to_scan as f64);
 
         let (partitioned_files, statistics) = self.partitioned_files(manifest_files);
         self.create_parquet_physical_plan(

src/stats.rs

@@ -137,13 +137,19 @@ pub async fn update_deleted_stats(
                 "parquet",
                 &manifest_date,
             ]);
-            let _ = TOTAL_EVENTS_INGESTED_DATE.remove_label_values(&["json", &manifest_date]);
-            let _ = TOTAL_EVENTS_INGESTED_SIZE_DATE.remove_label_values(&["json", &manifest_date]);
-            let _ =
-                TOTAL_EVENTS_STORAGE_SIZE_DATE.remove_label_values(&["parquet", &manifest_date]);
+
             num_row += manifest.events_ingested as i64;
             ingestion_size += manifest.ingestion_size as i64;
             storage_size += manifest.storage_size as i64;
+            TOTAL_EVENTS_INGESTED_DATE
+                .with_label_values(&["json", &manifest_date])
+                .sub(manifest.events_ingested as i64);
+            TOTAL_EVENTS_INGESTED_SIZE_DATE
+                .with_label_values(&["json", &manifest_date])
+                .sub(manifest.ingestion_size as i64);
+            TOTAL_EVENTS_STORAGE_SIZE_DATE
+                .with_label_values(&["parquet", &manifest_date])
+                .sub(manifest.storage_size as i64);
         }
     }
     EVENTS_DELETED