Collect metrics

Author: ladvoc

livekit-ffi/src/server/audio_stream.rs

@@ -13,12 +13,13 @@
 // limitations under the License.
 
 use std::borrow::Cow;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use futures_util::StreamExt;
 use livekit::track::Track;
 use livekit::webrtc::{audio_stream::native::NativeAudioStream, prelude::*};
 use livekit::{registered_audio_filter_plugin, AudioFilterAudioStream, AudioFilterStreamInfo};
+use metrics_logger::metrics::histogram;
 use tokio::sync::{broadcast, mpsc, oneshot};
 
 use super::audio_plugin::AudioStreamKind;
@@ -373,6 +374,9 @@ impl FfiAudioStream {
                     break;
                 }
                 frame = native_stream.next() => {
+
+                    let t0 = Instant::now();
+
                     let Some(frame) = frame else {
                         break;
                     };
@@ -447,6 +451,8 @@ impl FfiAudioStream {
                         }
                     }
 
+                    let delta = t0.elapsed();
+                    histogram!("forward_audio_frame").record(delta.as_millis() as f64);
                 }
             }
         }

livekit-ffi/src/server/mod.rs

@@ -15,19 +15,18 @@
 use std::{
     error::Error,
     sync::{
-        atomic::{AtomicU64, Ordering},
-        Arc,
+        Arc, atomic::{AtomicU64, Ordering}
     },
     thread,
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 use dashmap::{mapref::one::MappedRef, DashMap};
 use downcast_rs::{impl_downcast, Downcast};
 use livekit::webrtc::{
     native::apm::AudioProcessingModule, native::audio_resampler::AudioResampler, prelude::*,
 };
-use metrics_logger::{LogMode, MetricsLogger, metrics};
+use metrics_logger::{LogMode, MetricsLogger, metrics::{self, histogram}};
 use parking_lot::{deadlock, Mutex};
 use tokio::{sync::oneshot, task::JoinHandle};
 
@@ -173,13 +172,19 @@ impl FfiServer {
     }
 
     pub fn send_event(&self, message: proto::ffi_event::Message) -> FfiResult<()> {
+
+        let t0 = Instant::now();
+
         let cb = self
             .config
             .lock()
             .as_ref()
             .map_or_else(|| Err(FfiError::NotConfigured), |c| Ok(c.callback_fn.clone()))?;
 
         cb(proto::FfiEvent { message: Some(message) });
+
+        let delta = t0.elapsed();
+        histogram!("send_event").record(delta.as_millis() as f64);
         Ok(())
     }
 
@@ -197,7 +202,12 @@ impl FfiServer {
     where
         T: FfiHandle,
     {
+        let t0 = Instant::now();
+
         self.ffi_handles.insert(id, Box::new(handle));
+
+        let delta = t0.elapsed();
+        histogram!("store_handle").record(delta.as_millis() as f64);
     }
 
     pub fn retrieve_handle<T>(
@@ -211,6 +221,8 @@ impl FfiServer {
             return Err(FfiError::InvalidRequest("handle is invalid".into()));
         }
 
+        let t0 = Instant::now();
+
         let handle =
             self.ffi_handles.get(&id).ok_or(FfiError::InvalidRequest("handle not found".into()))?;
 
@@ -220,6 +232,9 @@ impl FfiServer {
             return Err(FfiError::InvalidRequest(msg.into()));
         }
 
+        let delta = t0.elapsed();
+        histogram!("retrieve_handle").record(delta.as_millis() as f64);
+
         let handle = handle.map(|v| v.downcast_ref::<T>().unwrap());
         Ok(handle)
     }
@@ -246,8 +261,15 @@ impl FfiServer {
     }
 
     pub fn drop_handle(&self, id: FfiHandleId) -> bool {
+
+        let t0 = Instant::now();
+
         let existed = self.ffi_handles.remove(&id).is_some();
         self.handle_dropped_txs.remove(&id);
+
+        let delta = t0.elapsed();
+        histogram!("drop_handle").record(delta.as_millis() as f64);
+
         return existed;
     }