Record how long it takes to flush writer buffers

Author: sandhose

crates/syn2mas/src/mas_writer/mod.rs

@@ -14,10 +14,12 @@ use std::{
         Arc,
         atomic::{AtomicU32, Ordering},
     },
+    time::Instant,
 };
 
 use chrono::{DateTime, Utc};
 use futures_util::{FutureExt, TryStreamExt, future::BoxFuture};
+use opentelemetry::KeyValue;
 use sqlx::{Executor, PgConnection, query, query_as};
 use thiserror::Error;
 use thiserror_ext::{Construct, ContextInto};
@@ -29,7 +31,7 @@ use self::{
     constraint_pausing::{ConstraintDescription, IndexDescription},
     locking::LockedMasDatabase,
 };
-use crate::Progress;
+use crate::{Progress, telemetry::WRITER_FLUSH_TIME};
 
 pub mod checks;
 pub mod locking;
@@ -672,6 +674,7 @@ impl MasWriter {
                         is_guests.push(is_guest);
                     }
 
+                    let start = Instant::now();
                     sqlx::query!(
                         r#"
                         INSERT INTO syn2mas__users (
@@ -698,6 +701,12 @@ impl MasWriter {
                     .await
                     .into_database("writing users to MAS")?;
 
+                    let elapsed = start.elapsed();
+                    WRITER_FLUSH_TIME.record(
+                        elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                        &[KeyValue::new("entity", "users")],
+                    );
+
                     Ok(())
                 })
             })
@@ -737,6 +746,7 @@ impl MasWriter {
                 versions.push(MIGRATED_PASSWORD_VERSION.into());
             }
 
+            let start = Instant::now();
             sqlx::query!(
                 r#"
                 INSERT INTO syn2mas__user_passwords
@@ -750,6 +760,12 @@ impl MasWriter {
                 &versions[..],
             ).execute(&mut *conn).await.into_database("writing users to MAS")?;
 
+            let elapsed = start.elapsed();
+            WRITER_FLUSH_TIME.record(
+                elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                &[KeyValue::new("entity", "user_passwords")],
+            );
+
             Ok(())
         })).boxed()
     }
@@ -779,6 +795,7 @@ impl MasWriter {
                     created_ats.push(created_at);
                 }
 
+                let start = Instant::now();
                 // `confirmed_at` is going to get removed in a future MAS release,
                 // so just populate with `created_at`
                 sqlx::query!(
@@ -793,6 +810,12 @@ impl MasWriter {
                     &created_ats[..],
                 ).execute(&mut *conn).await.into_database("writing emails to MAS")?;
 
+                let elapsed = start.elapsed();
+                WRITER_FLUSH_TIME.record(
+                    elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                    &[KeyValue::new("entity", "email_threepids")],
+                );
+
                 Ok(())
             })
         }).boxed()
@@ -823,6 +846,7 @@ impl MasWriter {
                     created_ats.push(created_at);
                 }
 
+                let start = Instant::now();
                 sqlx::query!(
                     r#"
                     INSERT INTO syn2mas__user_unsupported_third_party_ids
@@ -835,6 +859,12 @@ impl MasWriter {
                     &created_ats[..],
                 ).execute(&mut *conn).await.into_database("writing unsupported threepids to MAS")?;
 
+                let elapsed = start.elapsed();
+                WRITER_FLUSH_TIME.record(
+                    elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                    &[KeyValue::new("entity", "unsupported_threepids")],
+                );
+
                 Ok(())
             })
         }).boxed()
@@ -868,6 +898,7 @@ impl MasWriter {
                     created_ats.push(created_at);
                 }
 
+                let start = Instant::now();
                 sqlx::query!(
                     r#"
                     INSERT INTO syn2mas__upstream_oauth_links
@@ -881,6 +912,12 @@ impl MasWriter {
                     &created_ats[..],
                 ).execute(&mut *conn).await.into_database("writing unsupported threepids to MAS")?;
 
+                let elapsed = start.elapsed();
+                WRITER_FLUSH_TIME.record(
+                    elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                    &[KeyValue::new("entity", "upstream_oauth_links")],
+                );
+
                 Ok(())
             })
         }).boxed()
@@ -929,6 +966,7 @@ impl MasWriter {
                         user_agents.push(user_agent);
                     }
 
+                    let start = Instant::now();
                     sqlx::query!(
                         r#"
                         INSERT INTO syn2mas__compat_sessions (
@@ -959,6 +997,12 @@ impl MasWriter {
                     .await
                     .into_database("writing compat sessions to MAS")?;
 
+                    let elapsed = start.elapsed();
+                    WRITER_FLUSH_TIME.record(
+                        elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                        &[KeyValue::new("entity", "compat_sessions")],
+                    );
+
                     Ok(())
                 })
             })
@@ -995,6 +1039,7 @@ impl MasWriter {
                         expires_ats.push(expires_at);
                     }
 
+                    let start = Instant::now();
                     sqlx::query!(
                         r#"
                         INSERT INTO syn2mas__compat_access_tokens (
@@ -1021,6 +1066,12 @@ impl MasWriter {
                     .await
                     .into_database("writing compat access tokens to MAS")?;
 
+                    let elapsed = start.elapsed();
+                    WRITER_FLUSH_TIME.record(
+                        elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                        &[KeyValue::new("entity", "compat_access_tokens")],
+                    );
+
                     Ok(())
                 })
             })
@@ -1056,6 +1107,7 @@ impl MasWriter {
                         created_ats.push(created_at);
                     }
 
+                    let start = Instant::now();
                     sqlx::query!(
                         r#"
                         INSERT INTO syn2mas__compat_refresh_tokens (
@@ -1081,6 +1133,12 @@ impl MasWriter {
                     .await
                     .into_database("writing compat refresh tokens to MAS")?;
 
+                    let elapsed = start.elapsed();
+                    WRITER_FLUSH_TIME.record(
+                        elapsed.as_millis().try_into().unwrap_or(u64::MAX),
+                        &[KeyValue::new("entity", "compat_refresh_tokens")],
+                    );
+
                     Ok(())
                 })
             })

crates/syn2mas/src/telemetry.rs

@@ -2,7 +2,7 @@ use std::sync::LazyLock;
 
 use opentelemetry::{
     InstrumentationScope,
-    metrics::{Counter, Gauge, Meter},
+    metrics::{Counter, Gauge, Histogram, Meter},
 };
 use opentelemetry_semantic_conventions as semcov;
 
@@ -37,6 +37,14 @@ pub static SKIPPED_COUNTER: LazyLock<Counter<u64>> = LazyLock::new(|| {
         .build()
 });
 
+pub static WRITER_FLUSH_TIME: LazyLock<Histogram<u64>> = LazyLock::new(|| {
+    METER
+        .u64_histogram("syn2mas.writer.flush_time")
+        .with_description("Time spent flushing the writer")
+        .with_unit("ms")
+        .build()
+});
+
 /// Attribute key for syn2mas.entity metrics representing what entity.
 pub const K_ENTITY: &str = "entity";