diff --git a/services/appflowy-collaborate/src/group/group_init.rs b/services/appflowy-collaborate/src/group/group_init.rs index 2977a7b6..e515a79f 100644 --- a/services/appflowy-collaborate/src/group/group_init.rs +++ b/services/appflowy-collaborate/src/group/group_init.rs @@ -186,7 +186,8 @@ impl CollabGroup { } res = updates.next() => { match res { - Some(Ok((_message_id, update))) => { + Some(Ok((message_id, update))) => { + state.metrics.observe_collab_stream_latency(message_id.timestamp_ms); Self::handle_inbound_update(&state, update).await; }, Some(Err(err)) => { diff --git a/services/appflowy-collaborate/src/metrics.rs b/services/appflowy-collaborate/src/metrics.rs index dea64309..a3cfef24 100644 --- a/services/appflowy-collaborate/src/metrics.rs +++ b/services/appflowy-collaborate/src/metrics.rs @@ -1,3 +1,4 @@ +use chrono::Utc; use prometheus_client::metrics::counter::Counter; use prometheus_client::metrics::gauge::Gauge; use prometheus_client::metrics::histogram::Histogram; @@ -22,6 +23,8 @@ pub struct CollabRealtimeMetrics { pub(crate) collab_size: Histogram, /// How big is the collab (with history, after applying all updates). pub(crate) full_collab_size: Histogram, + /// How long does it take since collab update is send to a stream to be read from it. + pub(crate) collab_stream_latency: Histogram, } impl CollabRealtimeMetrics { @@ -54,6 +57,13 @@ impl CollabRealtimeMetrics { ] .into_iter(), ), + // collab update xadd-to-xread latency: 5ms, 50ms, 100ms, 500ms, 1s, 5s, 10s, 30s, 60s + collab_stream_latency: Histogram::new( + [ + 5.0, 50.0, 100.0, 500.0, 1000.0, 5000.0, 10000.0, 30000.0, 60000.0, + ] + .into_iter(), + ), load_collab_count: Default::default(), load_full_collab_count: Default::default(), } @@ -112,8 +122,22 @@ impl CollabRealtimeMetrics { "number of collab loads (with history)", metrics.load_full_collab_count.clone(), ); + realtime_registry.register( + "collab_stream_latency", + "latency since collab update is send to a stream to be read from it", + metrics.collab_stream_latency.clone(), + ); metrics } + + pub fn observe_collab_stream_latency(&self, message_id_timestamp: u64) { + let now = Utc::now().timestamp_millis() as u64; + if now > message_id_timestamp { + self + .collab_stream_latency + .observe((now - message_id_timestamp) as f64); + } + } } #[derive(Clone)]