feat: telemetry report for parquet metrics (#25425)

- added mechanism within PersistedFile to expose parquet file related
  metrics. The details are updated when new snapshot is generated and
  also when all snapshots are loaded when the process starts up
- at the point of creating the telemetry payload these parquet metrics
  are looked up before sending it to the server.

Closes: #25418

Author: praveen-influx

Cargo.lock

@@ -18,8 +18,11 @@ use influxdb3_server::{
 use influxdb3_telemetry::store::TelemetryStore;
 use influxdb3_wal::{Gen1Duration, WalConfig};
 use influxdb3_write::{
-    last_cache::LastCacheProvider, parquet_cache::create_cached_obj_store_and_oracle,
-    persister::Persister, write_buffer::WriteBufferImpl, WriteBuffer,
+    last_cache::LastCacheProvider,
+    parquet_cache::create_cached_obj_store_and_oracle,
+    persister::Persister,
+    write_buffer::{persisted_files::PersistedFiles, WriteBufferImpl},
+    WriteBuffer,
 };
 use iox_query::exec::{DedicatedExecutor, Executor, ExecutorConfig};
 use iox_time::SystemProvider;
@@ -415,29 +418,21 @@ pub async fn command(config: Config) -> Result<()> {
         snapshot_size: config.wal_snapshot_size,
     };
 
-    let catalog = persister
-        .load_or_create_catalog()
-        .await
-        .map_err(Error::InitializePersistedCatalog)?;
+    let catalog = Arc::new(
+        persister
+            .load_or_create_catalog()
+            .await
+            .map_err(Error::InitializePersistedCatalog)?,
+    );
 
     let last_cache = LastCacheProvider::new_from_catalog(&catalog.clone_inner())
         .map_err(Error::InitializeLastCache)?;
     info!(instance_id = ?catalog.instance_id(), "Catalog initialized with");
 
-    let telemetry_store =
-        setup_telemetry_store(&config.object_store_config, catalog.instance_id(), num_cpus).await;
-
-    let common_state = CommonServerState::new(
-        Arc::clone(&metrics),
-        trace_exporter,
-        trace_header_parser,
-        Arc::clone(&telemetry_store),
-    )?;
-
-    let write_buffer: Arc<dyn WriteBuffer> = Arc::new(
+    let write_buffer_impl = Arc::new(
         WriteBufferImpl::new(
             Arc::clone(&persister),
-            Arc::new(catalog),
+            Arc::clone(&catalog),
             Arc::new(last_cache),
             Arc::<SystemProvider>::clone(&time_provider),
             Arc::clone(&exec),
@@ -447,6 +442,24 @@ pub async fn command(config: Config) -> Result<()> {
         .await
         .map_err(|e| Error::WriteBufferInit(e.into()))?,
     );
+
+    let telemetry_store = setup_telemetry_store(
+        &config.object_store_config,
+        catalog.instance_id(),
+        num_cpus,
+        Arc::clone(&write_buffer_impl.persisted_files()),
+    )
+    .await;
+
+    let write_buffer: Arc<dyn WriteBuffer> = write_buffer_impl;
+
+    let common_state = CommonServerState::new(
+        Arc::clone(&metrics),
+        trace_exporter,
+        trace_header_parser,
+        Arc::clone(&telemetry_store),
+    )?;
+
     let query_executor = Arc::new(QueryExecutorImpl::new(
         write_buffer.catalog(),
         Arc::clone(&write_buffer),
@@ -486,6 +499,7 @@ async fn setup_telemetry_store(
     object_store_config: &ObjectStoreConfig,
     instance_id: Arc<str>,
     num_cpus: usize,
+    persisted_files: Arc<PersistedFiles>,
 ) -> Arc<TelemetryStore> {
     let os = std::env::consts::OS;
     let influxdb_pkg_version = env!("CARGO_PKG_VERSION");
@@ -503,6 +517,7 @@ async fn setup_telemetry_store(
         Arc::from(influx_version),
         Arc::from(storage_type),
         num_cpus,
+        persisted_files,
     )
     .await
 }

influxdb3/src/commands/serve.rs

@@ -748,14 +748,6 @@ mod tests {
     async fn setup_server(start_time: i64) -> (String, CancellationToken, Arc<dyn WriteBuffer>) {
         let trace_header_parser = trace_http::ctx::TraceHeaderParser::new();
         let metrics = Arc::new(metric::Registry::new());
-        let dummy_telem_store = TelemetryStore::new_without_background_runners();
-        let common_state = crate::CommonServerState::new(
-            Arc::clone(&metrics),
-            None,
-            trace_header_parser,
-            dummy_telem_store,
-        )
-        .unwrap();
         let object_store: Arc<DynObjectStore> = Arc::new(object_store::memory::InMemory::new());
         let time_provider = Arc::new(MockProvider::new(Time::from_timestamp_nanos(start_time)));
         let (object_store, parquet_cache) =
@@ -777,9 +769,7 @@ mod tests {
         let persister = Arc::new(Persister::new(Arc::clone(&object_store), "test_host"));
         let dummy_host_id = Arc::from("dummy-host-id");
         let instance_id = Arc::from("dummy-instance-id");
-        let telemetry_store = TelemetryStore::new_without_background_runners();
-
-        let write_buffer: Arc<dyn WriteBuffer> = Arc::new(
+        let write_buffer_impl = Arc::new(
             influxdb3_write::write_buffer::WriteBufferImpl::new(
                 Arc::clone(&persister),
                 Arc::new(Catalog::new(dummy_host_id, instance_id)),
@@ -792,6 +782,18 @@ mod tests {
             .await
             .unwrap(),
         );
+
+        let dummy_telem_store = TelemetryStore::new_without_background_runners(Arc::clone(
+            &write_buffer_impl.persisted_files(),
+        ));
+        let write_buffer: Arc<dyn WriteBuffer> = write_buffer_impl;
+        let common_state = crate::CommonServerState::new(
+            Arc::clone(&metrics),
+            None,
+            trace_header_parser,
+            Arc::clone(&dummy_telem_store),
+        )
+        .unwrap();
         let query_executor = crate::query_executor::QueryExecutorImpl::new(
             write_buffer.catalog(),
             Arc::clone(&write_buffer),
@@ -800,7 +802,7 @@ mod tests {
             Arc::new(HashMap::new()),
             10,
             10,
-            telemetry_store,
+            Arc::clone(&dummy_telem_store),
         );
 
         // bind to port 0 will assign a random available port:

influxdb3_server/src/lib.rs

@@ -645,7 +645,7 @@ mod tests {
         let executor = make_exec(Arc::clone(&object_store));
         let host_id = Arc::from("dummy-host-id");
         let instance_id = Arc::from("instance-id");
-        let write_buffer: Arc<dyn WriteBuffer> = Arc::new(
+        let write_buffer_impl = Arc::new(
             WriteBufferImpl::new(
                 Arc::clone(&persister),
                 Arc::new(Catalog::new(host_id, instance_id)),
@@ -663,9 +663,13 @@ mod tests {
             .await
             .unwrap(),
         );
+
+        let dummy_telem_store = TelemetryStore::new_without_background_runners(Arc::clone(
+            &write_buffer_impl.persisted_files(),
+        ));
+        let write_buffer: Arc<dyn WriteBuffer> = write_buffer_impl;
         let metrics = Arc::new(Registry::new());
         let df_config = Arc::new(Default::default());
-        let dummy_telem_store = TelemetryStore::new_without_background_runners();
         let query_executor = QueryExecutorImpl::new(
             write_buffer.catalog(),
             Arc::clone(&write_buffer),

influxdb3_server/src/query_executor.rs

@@ -18,6 +18,9 @@ sysinfo.workspace = true
 num.workspace = true
 thiserror.workspace = true
 
+# Local Deps
+influxdb3_write = { path = "../influxdb3_write" }
+
 [dev-dependencies]
 test-log.workspace = true
 proptest.workspace = true

influxdb3_telemetry/Cargo.toml

@@ -62,6 +62,10 @@ pub(crate) struct TelemetryPayload {
     pub query_requests_min: u64,
     pub query_requests_max: u64,
     pub query_requests_avg: u64,
+    // parquet files
+    pub parquet_file_count: u64,
+    pub parquet_file_size_mb: f64,
+    pub parquet_row_count: u64,
 }
 
 pub(crate) async fn send_telemetry_in_background(

influxdb3_telemetry/src/sender.rs

@@ -1,5 +1,6 @@
 use std::{sync::Arc, time::Duration};
 
+use influxdb3_write::write_buffer::persisted_files::PersistedFiles;
 use num::Float;
 use observability_deps::tracing::{debug, warn};
 
@@ -15,6 +16,7 @@ use crate::{
 #[derive(Debug)]
 pub struct TelemetryStore {
     inner: parking_lot::Mutex<TelemetryStoreInner>,
+    persisted_files: Arc<PersistedFiles>,
 }
 
 const SAMPLER_INTERVAL_SECS: u64 = 60;
@@ -27,6 +29,7 @@ impl TelemetryStore {
         influx_version: Arc<str>,
         storage_type: Arc<str>,
         cores: usize,
+        persisted_files: Arc<PersistedFiles>,
     ) -> Arc<Self> {
         debug!(
             instance_id = ?instance_id,
@@ -39,6 +42,7 @@ impl TelemetryStore {
         let inner = TelemetryStoreInner::new(instance_id, os, influx_version, storage_type, cores);
         let store = Arc::new(TelemetryStore {
             inner: parking_lot::Mutex::new(inner),
+            persisted_files,
         });
 
         if !cfg!(test) {
@@ -52,7 +56,7 @@ impl TelemetryStore {
         store
     }
 
-    pub fn new_without_background_runners() -> Arc<Self> {
+    pub fn new_without_background_runners(persisted_files: Arc<PersistedFiles>) -> Arc<Self> {
         let instance_id = Arc::from("dummy-instance-id");
         let os = Arc::from("Linux");
         let influx_version = Arc::from("influxdb3-0.1.0");
@@ -61,6 +65,7 @@ impl TelemetryStore {
         let inner = TelemetryStoreInner::new(instance_id, os, influx_version, storage_type, cores);
         Arc::new(TelemetryStore {
             inner: parking_lot::Mutex::new(inner),
+            persisted_files,
         })
     }
 
@@ -100,7 +105,12 @@ impl TelemetryStore {
 
     pub(crate) fn snapshot(&self) -> TelemetryPayload {
         let inner_store = self.inner.lock();
-        inner_store.snapshot()
+        let (file_count, size_mb, row_count) = self.persisted_files.get_metrics();
+        let mut payload = inner_store.snapshot();
+        payload.parquet_file_count = file_count;
+        payload.parquet_file_size_mb = size_mb;
+        payload.parquet_row_count = row_count;
+        payload
     }
 }
 
@@ -194,6 +204,11 @@ impl TelemetryStoreInner {
             query_requests_min: self.reads.num_queries.min,
             query_requests_max: self.reads.num_queries.max,
             query_requests_avg: self.reads.num_queries.avg,
+
+            // hmmm. will be nice to pass persisted file in?
+            parquet_file_count: 0,
+            parquet_file_size_mb: 0.0,
+            parquet_row_count: 0,
         }
     }
 
@@ -265,13 +280,18 @@ mod tests {
 
     #[test_log::test(tokio::test)]
     async fn test_telemetry_store_cpu_mem() {
+        let persisted_snapshots = Vec::new();
         // create store
+        let persisted_files = Arc::from(PersistedFiles::new_from_persisted_snapshots(
+            persisted_snapshots,
+        ));
         let store: Arc<TelemetryStore> = TelemetryStore::new(
             Arc::from("some-instance-id"),
             Arc::from("Linux"),
             Arc::from("OSS-v3.0"),
             Arc::from("Memory"),
             10,
+            persisted_files,
         )
         .await;
         // check snapshot
@@ -301,6 +321,9 @@ mod tests {
         assert_eq!(expected_mem_in_mb, snapshot.memory_used_mb_min);
         assert_eq!(128, snapshot.memory_used_mb_max);
         assert_eq!(122, snapshot.memory_used_mb_avg);
+        assert_eq!(0, snapshot.parquet_file_count);
+        assert_eq!(0.0, snapshot.parquet_file_size_mb);
+        assert_eq!(0, snapshot.parquet_row_count);
 
         // add some writes
         store.add_write_metrics(100, 100);