feat: add trigger count to telemetry (#26426)

* feat: add trigger count to telemetry

closes: #26285

* refactor: do trigger counts by type

Author: praveen-influx

Cargo.lock

@@ -36,7 +36,10 @@ use influxdb3_server::{
 };
 use influxdb3_shutdown::{ShutdownManager, wait_for_signal};
 use influxdb3_sys_events::SysEventStore;
-use influxdb3_telemetry::store::{CreateTelemetryStoreArgs, TelemetryStore};
+use influxdb3_telemetry::{
+    ProcessingEngineMetrics,
+    store::{CreateTelemetryStoreArgs, TelemetryStore},
+};
 use influxdb3_wal::{Gen1Duration, WalConfig};
 use influxdb3_write::{
     WriteBuffer,
@@ -694,15 +697,16 @@ pub async fn command(config: Config) -> Result<()> {
     .await;
 
     info!("setting up telemetry store");
-    let telemetry_store = setup_telemetry_store(
-        &config.object_store_config,
-        node_def.instance_id(),
+    let telemetry_store = setup_telemetry_store(TelemetryStoreSetupArgs {
+        object_store_config: &config.object_store_config,
+        instance_id: node_def.instance_id(),
         num_cpus,
-        Some(Arc::clone(&write_buffer_impl.persisted_files())),
-        config.telemetry_endpoint.as_str(),
-        config.disable_telemetry_upload,
-        catalog.catalog_uuid().to_string(),
-    )
+        persisted_files: Some(Arc::clone(&write_buffer_impl.persisted_files())),
+        telemetry_endpoint: &config.telemetry_endpoint,
+        disable_upload: config.disable_telemetry_upload,
+        catalog_uuid: catalog.catalog_uuid().to_string(),
+        processing_engine_metrics: Arc::clone(&catalog) as Arc<dyn ProcessingEngineMetrics>,
+    })
     .await;
 
     let write_buffer: Arc<dyn WriteBuffer> = write_buffer_impl;
@@ -916,14 +920,28 @@ fn determine_package_manager() -> Arc<dyn PythonEnvironmentManager> {
     Arc::new(DisabledManager)
 }
 
-async fn setup_telemetry_store(
-    object_store_config: &ObjectStoreConfig,
+struct TelemetryStoreSetupArgs<'a> {
+    object_store_config: &'a ObjectStoreConfig,
     instance_id: Arc<str>,
     num_cpus: usize,
     persisted_files: Option<Arc<PersistedFiles>>,
-    telemetry_endpoint: &str,
+    telemetry_endpoint: &'a str,
     disable_upload: bool,
     catalog_uuid: String,
+    processing_engine_metrics: Arc<dyn ProcessingEngineMetrics>,
+}
+
+async fn setup_telemetry_store(
+    TelemetryStoreSetupArgs {
+        object_store_config,
+        instance_id,
+        num_cpus,
+        persisted_files,
+        telemetry_endpoint,
+        disable_upload,
+        catalog_uuid,
+        processing_engine_metrics,
+    }: TelemetryStoreSetupArgs<'_>,
 ) -> Arc<TelemetryStore> {
     let os = std::env::consts::OS;
     let influxdb_pkg_version = env!("CARGO_PKG_VERSION");
@@ -937,7 +955,10 @@ async fn setup_telemetry_store(
 
     if disable_upload {
         debug!("Initializing TelemetryStore with upload disabled.");
-        TelemetryStore::new_without_background_runners(persisted_files.map(|p| p as _))
+        TelemetryStore::new_without_background_runners(
+            persisted_files.map(|p| p as _),
+            processing_engine_metrics,
+        )
     } else {
         debug!("Initializing TelemetryStore with upload enabled for {telemetry_endpoint}.");
         TelemetryStore::new(CreateTelemetryStoreArgs {
@@ -949,6 +970,7 @@ async fn setup_telemetry_store(
             persisted_files: persisted_files.map(|p| p as _),
             telemetry_endpoint: telemetry_endpoint.to_string(),
             catalog_uuid,
+            processing_engine_metrics,
         })
         .await
     }

influxdb3/src/commands/serve.rs

@@ -19,6 +19,7 @@ influxdb3_id = { path = "../influxdb3_id" }
 influxdb3_process = { path = "../influxdb3_process" }
 influxdb3_shutdown = { path = "../influxdb3_shutdown" }
 influxdb3_wal = { path = "../influxdb3_wal" }
+influxdb3_telemetry = { path = "../influxdb3_telemetry" }
 
 # crates.io dependencies
 anyhow.workspace = true

influxdb3_catalog/Cargo.toml

@@ -14,6 +14,7 @@ use influxdb3_id::{
     TriggerId,
 };
 use influxdb3_shutdown::ShutdownToken;
+use influxdb3_telemetry::ProcessingEngineMetrics;
 use iox_time::{Time, TimeProvider};
 use metric::Registry;
 use metrics::CatalogMetrics;
@@ -41,6 +42,7 @@ pub use update::{CatalogUpdate, DatabaseCatalogTransaction, Prompt};
 
 use crate::channel::{CatalogSubscriptions, CatalogUpdateReceiver};
 use crate::log::CreateAdminTokenDetails;
+use crate::log::TriggerSpecificationDefinition;
 use crate::log::{
     CreateDatabaseLog, DatabaseBatch, DatabaseCatalogOp, NodeBatch, NodeCatalogOp, NodeMode,
     RegenerateAdminTokenDetails, RegisterNodeLog, StopNodeLog, TokenBatch, TokenCatalogOp,
@@ -615,6 +617,12 @@ impl TokenProvider for Catalog {
     }
 }
 
+impl ProcessingEngineMetrics for Catalog {
+    fn num_triggers(&self) -> (u64, u64, u64, u64) {
+        self.inner.read().num_triggers()
+    }
+}
+
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct Repository<I: CatalogId, R: CatalogResource> {
     pub(crate) repo: SerdeVecMap<I, Arc<R>>,
@@ -954,6 +962,33 @@ impl InnerCatalog {
     pub fn db_exists(&self, db_id: DbId) -> bool {
         self.databases.get_by_id(&db_id).is_some()
     }
+
+    pub fn num_triggers(&self) -> (u64, u64, u64, u64) {
+        self.databases
+            .iter()
+            .map(|(_, db)| db.trigger_count_by_type())
+            .fold(
+                (0, 0, 0, 0),
+                |(
+                    mut overall_wal_count,
+                    mut overall_all_wal_count,
+                    mut overall_schedule_count,
+                    mut overall_request_count,
+                ),
+                 (wal_count, all_wal_count, schedule_count, request_count)| {
+                    overall_wal_count += wal_count;
+                    overall_all_wal_count += all_wal_count;
+                    overall_schedule_count += schedule_count;
+                    overall_request_count += request_count;
+                    (
+                        overall_wal_count,
+                        overall_all_wal_count,
+                        overall_schedule_count,
+                        overall_request_count,
+                    )
+                },
+            )
+    }
 }
 
 #[derive(Debug, Eq, PartialEq, Clone)]
@@ -1162,6 +1197,26 @@ impl DatabaseSchema {
             .cloned()
             .collect()
     }
+
+    pub fn trigger_count_by_type(&self) -> (u64, u64, u64, u64) {
+        self.processing_engine_triggers.iter().fold(
+            (0, 0, 0, 0),
+            |(mut wal_count, mut all_wal_count, mut schedule_count, mut request_count),
+             (_, trigger)| {
+                match trigger.trigger {
+                    // wal
+                    TriggerSpecificationDefinition::SingleTableWalWrite { .. } => wal_count += 1,
+                    TriggerSpecificationDefinition::AllTablesWalWrite => all_wal_count += 1,
+                    // schedule
+                    TriggerSpecificationDefinition::Schedule { .. }
+                    | TriggerSpecificationDefinition::Every { .. } => schedule_count += 1,
+                    // request
+                    TriggerSpecificationDefinition::RequestPath { .. } => request_count += 1,
+                };
+                (wal_count, all_wal_count, schedule_count, request_count)
+            },
+        )
+    }
 }
 
 trait UpdateDatabaseSchema {

influxdb3_catalog/src/catalog.rs

@@ -286,9 +286,9 @@ mod tests {
     use influxdb3_sys_events::SysEventStore;
     use influxdb3_telemetry::store::TelemetryStore;
     use influxdb3_wal::WalConfig;
-    use influxdb3_write::WriteBuffer;
     use influxdb3_write::persister::Persister;
     use influxdb3_write::write_buffer::persisted_files::PersistedFiles;
+    use influxdb3_write::{Bufferer, WriteBuffer};
     use iox_query::exec::{DedicatedExecutor, Executor, ExecutorConfig};
     use iox_time::{MockProvider, Time};
     use object_store::DynObjectStore;
@@ -880,8 +880,13 @@ mod tests {
         let sys_events_store = Arc::new(SysEventStore::new(Arc::clone(&time_provider) as _));
         let parquet_metrics_provider: Arc<PersistedFiles> =
             Arc::clone(&write_buffer_impl.persisted_files());
-        let sample_telem_store =
-            TelemetryStore::new_without_background_runners(Some(parquet_metrics_provider));
+        let processing_engine_metrics_provider: Arc<Catalog> =
+            Arc::clone(&write_buffer_impl.catalog());
+
+        let sample_telem_store = TelemetryStore::new_without_background_runners(
+            Some(parquet_metrics_provider),
+            processing_engine_metrics_provider,
+        );
         let write_buffer: Arc<dyn WriteBuffer> = write_buffer_impl;
         let common_state = crate::CommonServerState::new(
             Arc::clone(&metrics),

influxdb3_server/src/lib.rs

@@ -763,7 +763,7 @@ mod tests {
     use influxdb3_telemetry::store::TelemetryStore;
     use influxdb3_wal::{Gen1Duration, WalConfig};
     use influxdb3_write::{
-        WriteBuffer,
+        Bufferer, WriteBuffer,
         persister::Persister,
         write_buffer::{WriteBufferImpl, WriteBufferImplArgs, persisted_files::PersistedFiles},
     };
@@ -864,7 +864,12 @@ mod tests {
         .unwrap();
 
         let persisted_files: Arc<PersistedFiles> = Arc::clone(&write_buffer_impl.persisted_files());
-        let telemetry_store = TelemetryStore::new_without_background_runners(Some(persisted_files));
+        let processing_engine_metrics_provider: Arc<Catalog> =
+            Arc::clone(&write_buffer_impl.catalog());
+        let telemetry_store = TelemetryStore::new_without_background_runners(
+            Some(persisted_files),
+            processing_engine_metrics_provider,
+        );
         let sys_events_store = Arc::new(SysEventStore::new(Arc::<MockProvider>::clone(
             &time_provider,
         )));

influxdb3_server/src/query_executor/mod.rs

@@ -24,3 +24,7 @@ pub type Result<T, E = TelemetryError> = std::result::Result<T, E>;
 pub trait ParquetMetrics: Send + Sync + std::fmt::Debug + 'static {
     fn get_metrics(&self) -> (u64, f64, u64);
 }
+
+pub trait ProcessingEngineMetrics: Send + Sync + std::fmt::Debug + 'static {
+    fn num_triggers(&self) -> (u64, u64, u64, u64);
+}

influxdb3_telemetry/src/lib.rs

@@ -109,24 +109,32 @@ fn sample_all_metrics(sampler: &mut CpuAndMemorySampler, store: &Arc<TelemetrySt
 #[cfg(test)]
 mod tests {
 
-    use crate::ParquetMetrics;
+    use crate::{ParquetMetrics, ProcessingEngineMetrics};
 
     use super::*;
 
     #[derive(Debug)]
-    struct MockParquetMetrics;
+    struct MockMetrics;
 
-    impl ParquetMetrics for MockParquetMetrics {
+    impl ParquetMetrics for MockMetrics {
         fn get_metrics(&self) -> (u64, f64, u64) {
             (10, 20.0, 30)
         }
     }
 
+    impl ProcessingEngineMetrics for MockMetrics {
+        fn num_triggers(&self) -> (u64, u64, u64, u64) {
+            (100, 110, 150, 200)
+        }
+    }
+
     #[test]
     fn test_sample_all_metrics() {
         let mut mock_sys_info_provider = MockSystemInfoProvider::new();
-        let store =
-            TelemetryStore::new_without_background_runners(Some(Arc::from(MockParquetMetrics)));
+        let store = TelemetryStore::new_without_background_runners(
+            Some(Arc::from(MockMetrics)),
+            Arc::from(MockMetrics) as Arc<dyn ProcessingEngineMetrics>,
+        );
 
         mock_sys_info_provider
             .expect_get_pid()
@@ -146,8 +154,10 @@ mod tests {
     #[test]
     fn test_sample_all_metrics_with_call_failure() {
         let mut mock_sys_info_provider = MockSystemInfoProvider::new();
-        let store =
-            TelemetryStore::new_without_background_runners(Some(Arc::from(MockParquetMetrics)));
+        let store = TelemetryStore::new_without_background_runners(
+            Some(Arc::from(MockMetrics)),
+            Arc::from(MockMetrics) as Arc<dyn ProcessingEngineMetrics>,
+        );
 
         mock_sys_info_provider
             .expect_get_pid()

influxdb3_telemetry/src/sampler.rs

@@ -87,6 +87,11 @@ pub(crate) struct TelemetryPayload {
     pub parquet_file_count: u64,
     pub parquet_file_size_mb: f64,
     pub parquet_row_count: u64,
+    // triggers (processing engine)
+    pub wal_single_triggers_count: u64,
+    pub wal_all_triggers_count: u64,
+    pub schedule_triggers_count: u64,
+    pub request_triggers_count: u64,
 }
 
 /// This function runs in the background and if any call fails
@@ -197,6 +202,10 @@ mod tests {
             write_lines_sum_1h: 200,
             write_mb_sum_1h: 200,
             query_requests_sum_1h: 200,
+            wal_single_triggers_count: 100,
+            wal_all_triggers_count: 100,
+            schedule_triggers_count: 150,
+            request_triggers_count: 155,
         }
     }
 }