Refactor metrics to event-based streaming with operation IDs

Author: gotocoding-DB

kernel/src/lib.rs

@@ -713,7 +713,7 @@ pub trait Engine: AsAny {
 
     /// Get the connector provided [`metrics::MetricsReporter`] for metrics collection.
     ///
-    /// Returns an optional reporter that will be notified of metrics from Delta operations.
+    /// Returns an optional reporter that will receive metric events from Delta operations.
     /// The default implementation returns None (no metrics reporting).
     fn get_metrics_reporter(&self) -> Option<Arc<dyn crate::metrics::MetricsReporter>> {
         None

kernel/src/metrics/mod.rs

@@ -2,25 +2,37 @@
 //!
 //! This module provides metrics tracking for various Delta operations including
 //! snapshot creation, scans, and transactions. Metrics are collected during operations
-//! and can be reported via the `MetricsReporter` trait.
+//! and reported as events via the `MetricsReporter` trait.
+//!
+//! Each operation (Snapshot, Transaction, Scan) is assigned a unique operation ID (UUID)
+//! when it starts, and all subsequent events for that operation reference this ID.
+//! This allows reporters to correlate events and track operation lifecycles.
 //!
 //! # Example: Implementing a Custom MetricsReporter
 //!
 //! ```
 //! use std::sync::Arc;
-//! use delta_kernel::metrics::{MetricsReporter, SnapshotMetrics};
+//! use delta_kernel::metrics::{MetricsReporter, MetricEvent};
 //!
 //! struct LoggingReporter;
 //!
 //! impl MetricsReporter for LoggingReporter {
-//!     fn report_snapshot(&self, metrics: &SnapshotMetrics) {
-//!         println!("Snapshot creation took {:?}", metrics.load_snapshot_total_duration);
-//!         println!("  - Protocol/Metadata load: {:?}", metrics.load_protocol_metadata_duration);
-//!         println!("  - Log segment load: {:?}", metrics.load_log_segment_duration);
-//!         println!("  - Files: {} commits, {} checkpoints, {} compactions",
-//!                  metrics.num_commit_files,
-//!                  metrics.num_checkpoint_files,
-//!                  metrics.num_compaction_files);
+//!     fn report(&self, event: MetricEvent) {
+//!         match event {
+//!             MetricEvent::SnapshotStarted { operation_id, table_path } => {
+//!                 println!("Snapshot started: {} for table {}", operation_id, table_path);
+//!             }
+//!             MetricEvent::LogSegmentLoaded { operation_id, duration, num_commit_files, .. } => {
+//!                 println!("  Log segment loaded in {:?}: {} commits", duration, num_commit_files);
+//!             }
+//!             MetricEvent::SnapshotCompleted { operation_id, version, total_duration } => {
+//!                 println!("Snapshot completed: v{} in {:?}", version, total_duration);
+//!             }
+//!             MetricEvent::SnapshotFailed { operation_id } => {
+//!                 println!("Snapshot failed: {}", operation_id);
+//!             }
+//!             _ => {}
+//!         }
 //!     }
 //! }
 //! ```
@@ -31,62 +43,79 @@
 //!
 //! ```
 //! use std::sync::Arc;
-//! use delta_kernel::metrics::{MetricsReporter, SnapshotMetrics};
+//! use delta_kernel::metrics::{MetricsReporter, MetricEvent};
 //!
 //! struct CompositeReporter {
 //!     reporters: Vec<Arc<dyn MetricsReporter>>,
 //! }
 //!
 //! impl MetricsReporter for CompositeReporter {
-//!     fn report_snapshot(&self, metrics: &SnapshotMetrics) {
+//!     fn report(&self, event: MetricEvent) {
 //!         for reporter in &self.reporters {
-//!             reporter.report_snapshot(metrics);
+//!             reporter.report(event.clone());
 //!         }
 //!     }
 //! }
 //! ```
 
 use std::time::{Duration, Instant};
+use uuid::Uuid;
 
-/// Trait for reporting metrics from Delta operations.
+/// Trait for reporting metrics events from Delta operations.
 ///
-/// Implementations of this trait receive metrics after operations complete
+/// Implementations of this trait receive metric events as they occur during operations
 /// and can forward them to monitoring systems like Prometheus, DataDog, etc.
+///
+/// Events are emitted throughout an operation's lifecycle, allowing real-time monitoring.
 pub trait MetricsReporter: Send + Sync {
-    /// Report snapshot creation metrics.
-    fn report_snapshot(&self, metrics: &SnapshotMetrics);
+    /// Report a metric event.
+    fn report(&self, event: MetricEvent);
 }
 
-/// Metrics collected during snapshot creation.
+/// Metric events emitted during Delta Kernel operations.
 ///
-/// These metrics track the time spent in various phases of loading a snapshot,
-/// including log segment construction and protocol/metadata loading.
-/// They also track counts of files processed during snapshot creation.
-#[derive(Debug, Clone, Default)]
-pub struct SnapshotMetrics {
-    /// Total time spent loading the snapshot (all operations combined).
-    pub load_snapshot_total_duration: Duration,
+/// Each event includes an `operation_id` (UUID) that uniquely identifies the operation
+/// instance. This allows correlating multiple events from the same operation.
+#[derive(Debug, Clone)]
+pub enum MetricEvent {
+    /// A snapshot creation operation has started.
+    SnapshotStarted {
+        operation_id: Uuid,
+        table_path: String,
+    },
 
-    /// Time spent loading protocol and metadata actions.
-    pub load_protocol_metadata_duration: Duration,
+    /// Log segment loading completed (listing and organizing log files).
+    LogSegmentLoaded {
+        operation_id: Uuid,
+        duration: Duration,
+        num_commit_files: u64,
+        num_checkpoint_files: u64,
+        num_compaction_files: u64,
+    },
 
-    /// Time spent building the log segment (listing and organizing log files).
-    pub load_log_segment_duration: Duration,
+    /// Protocol and metadata loading completed.
+    ProtocolMetadataLoaded {
+        operation_id: Uuid,
+        duration: Duration,
+    },
 
-    /// Number of commit files processed during snapshot creation.
-    pub num_commit_files: u64,
+    /// Snapshot creation completed successfully.
+    SnapshotCompleted {
+        operation_id: Uuid,
+        version: u64,
+        total_duration: Duration,
+    },
 
-    /// Number of checkpoint files processed during snapshot creation.
-    pub num_checkpoint_files: u64,
-
-    /// Number of compaction files processed during snapshot creation.
-    pub num_compaction_files: u64,
+    /// Snapshot creation failed.
+    SnapshotFailed { operation_id: Uuid },
 }
 
 /// A simple timer for tracking operation durations.
 ///
 /// # Example
-/// ```ignore
+/// ```
+/// use delta_kernel::metrics::Timer;
+///
 /// let timer = Timer::new();
 /// // ... do work ...
 /// let duration = timer.elapsed();

kernel/src/snapshot.rs

@@ -11,7 +11,7 @@ use crate::checkpoint::CheckpointWriter;
 use crate::committer::Committer;
 use crate::listed_log_files::ListedLogFiles;
 use crate::log_segment::LogSegment;
-use crate::metrics::{SnapshotMetrics, Timer};
+use crate::metrics::{MetricEvent, Timer};
 use crate::path::ParsedLogPath;
 use crate::scan::ScanBuilder;
 use crate::schema::SchemaRef;
@@ -21,6 +21,7 @@ use crate::transaction::Transaction;
 use crate::LogCompactionWriter;
 use crate::{DeltaResult, Engine, Error, Version};
 use delta_kernel_derive::internal_api;
+use uuid::Uuid;
 
 mod builder;
 pub use builder::SnapshotBuilder;
@@ -255,51 +256,61 @@ impl Snapshot {
         log_segment: LogSegment,
         engine: &dyn Engine,
     ) -> DeltaResult<Self> {
-        Self::try_new_from_log_segment_with_metrics(
-            location,
-            log_segment,
-            engine,
-            std::time::Duration::default(),
-        )
+        Self::try_new_from_log_segment_with_metrics(location, log_segment, engine, None)
     }
 
     pub(crate) fn try_new_from_log_segment_with_metrics(
         location: Url,
         log_segment: LogSegment,
         engine: &dyn Engine,
-        load_log_segment_duration: std::time::Duration,
+        operation_id: Option<Uuid>,
     ) -> DeltaResult<Self> {
+        let operation_id = operation_id.unwrap_or_else(Uuid::new_v4);
+        let reporter = engine.get_metrics_reporter();
         let timer = Timer::new();
 
-        let (metadata, protocol) = log_segment.read_metadata(engine)?;
-        let load_protocol_metadata_duration = timer.elapsed();
-
-        let table_configuration =
-            TableConfiguration::try_new(metadata, protocol, location, log_segment.end_version)?;
+        let result = (|| {
+            let (metadata, protocol) = log_segment.read_metadata(engine)?;
+            let protocol_metadata_duration = timer.elapsed();
 
-        let num_commit_files = log_segment.ascending_commit_files.len() as u64;
-        let num_checkpoint_files = log_segment.checkpoint_parts.len() as u64;
-        let num_compaction_files = log_segment.ascending_compaction_files.len() as u64;
-
-        let load_snapshot_total_duration = timer.elapsed();
+            if let Some(ref r) = reporter {
+                r.report(MetricEvent::ProtocolMetadataLoaded {
+                    operation_id,
+                    duration: protocol_metadata_duration,
+                });
+            }
 
-        let metrics = SnapshotMetrics {
-            load_snapshot_total_duration,
-            load_protocol_metadata_duration,
-            load_log_segment_duration,
-            num_commit_files,
-            num_checkpoint_files,
-            num_compaction_files,
-        };
+            let table_configuration = TableConfiguration::try_new(
+                metadata,
+                protocol,
+                location,
+                log_segment.end_version,
+            )?;
 
-        if let Some(reporter) = engine.get_metrics_reporter() {
-            reporter.report_snapshot(&metrics);
+            Ok(Self {
+                log_segment,
+                table_configuration,
+            })
+        })();
+
+        match result {
+            Ok(snapshot) => {
+                if let Some(ref r) = reporter {
+                    r.report(MetricEvent::SnapshotCompleted {
+                        operation_id,
+                        version: snapshot.version(),
+                        total_duration: timer.elapsed(),
+                    });
+                }
+                Ok(snapshot)
+            }
+            Err(e) => {
+                if let Some(ref r) = reporter {
+                    r.report(MetricEvent::SnapshotFailed { operation_id });
+                }
+                Err(e)
+            }
         }
-
-        Ok(Self {
-            log_segment,
-            table_configuration,
-        })
     }
 
     /// Creates a [`CheckpointWriter`] for generating a checkpoint from this snapshot.

kernel/src/snapshot/builder.rs

@@ -1,9 +1,10 @@
 //! Builder for creating [`Snapshot`] instances.
 use crate::log_path::LogPath;
 use crate::log_segment::LogSegment;
-use crate::metrics::Timer;
+use crate::metrics::{MetricEvent, Timer};
 use crate::snapshot::SnapshotRef;
 use crate::{DeltaResult, Engine, Error, Snapshot, Version};
+use uuid::Uuid;
 
 use url::Url;
 
@@ -84,20 +85,51 @@ impl SnapshotBuilder {
     pub fn build(self, engine: &dyn Engine) -> DeltaResult<SnapshotRef> {
         let log_tail = self.log_tail.into_iter().map(Into::into).collect();
         if let Some(table_root) = self.table_root {
+            let operation_id = Uuid::new_v4();
+            let reporter = engine.get_metrics_reporter();
+
+            if let Some(ref r) = reporter {
+                r.report(MetricEvent::SnapshotStarted {
+                    operation_id,
+                    table_path: table_root.to_string(),
+                });
+            }
+
             let timer = Timer::new();
-            let log_segment = LogSegment::for_snapshot(
+            let log_segment_result = LogSegment::for_snapshot(
                 engine.storage_handler().as_ref(),
                 table_root.join("_delta_log/")?,
                 log_tail,
                 self.version,
-            )?;
-            let load_log_segment_duration = timer.elapsed();
+            );
+
+            let log_segment = match log_segment_result {
+                Ok(seg) => {
+                    let duration = timer.elapsed();
+                    if let Some(ref r) = reporter {
+                        r.report(MetricEvent::LogSegmentLoaded {
+                            operation_id,
+                            duration,
+                            num_commit_files: seg.ascending_commit_files.len() as u64,
+                            num_checkpoint_files: seg.checkpoint_parts.len() as u64,
+                            num_compaction_files: seg.ascending_compaction_files.len() as u64,
+                        });
+                    }
+                    seg
+                }
+                Err(e) => {
+                    if let Some(ref r) = reporter {
+                        r.report(MetricEvent::SnapshotFailed { operation_id });
+                    }
+                    return Err(e);
+                }
+            };
 
             Ok(Snapshot::try_new_from_log_segment_with_metrics(
                 table_root,
                 log_segment,
                 engine,
-                load_log_segment_duration,
+                Some(operation_id),
             )?
             .into())
         } else {