syn2mas: Add progress reporting to log and to opentelemetry metrics (#4215)

Author: reivilibre

Cargo.lock

@@ -61,6 +61,9 @@ syn2mas = { path = "./crates/syn2mas", version = "=0.14.1" }
 version = "0.14.1"
 features = ["axum", "axum-extra", "axum-json", "axum-query", "macros"]
 
+[workspace.dependencies.arc-swap]
+version = "1.7.1"
+
 # GraphQL server
 [workspace.dependencies.async-graphql]
 version = "7.0.15"

Cargo.toml

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, process::ExitCode};
+use std::{collections::HashMap, process::ExitCode, sync::atomic::Ordering, time::Duration};
 
 use anyhow::Context;
 use camino::Utf8PathBuf;
@@ -12,8 +12,10 @@ use mas_storage::SystemClock;
 use mas_storage_pg::MIGRATOR;
 use rand::thread_rng;
 use sqlx::{Connection, Either, PgConnection, postgres::PgConnectOptions, types::Uuid};
-use syn2mas::{LockedMasDatabase, MasWriter, SynapseReader, synapse_config};
-use tracing::{Instrument, error, info_span, warn};
+use syn2mas::{
+    LockedMasDatabase, MasWriter, Progress, ProgressStage, SynapseReader, synapse_config,
+};
+use tracing::{Instrument, error, info, info_span, warn};
 
 use crate::util::{DatabaseConnectOptions, database_connection_from_config_with_options};
 
@@ -248,7 +250,11 @@ impl Options {
                 #[allow(clippy::disallowed_methods)]
                 let mut rng = thread_rng();
 
-                // TODO progress reporting
+                let progress = Progress::default();
+
+                let occasional_progress_logger_task =
+                    tokio::spawn(occasional_progress_logger(progress.clone()));
+
                 let mas_matrix = MatrixConfig::extract(figment)?;
                 eprintln!("\n\n");
                 syn2mas::migrate(
@@ -258,11 +264,45 @@ impl Options {
                     &clock,
                     &mut rng,
                     provider_id_mappings,
+                    &progress,
                 )
                 .await?;
 
+                occasional_progress_logger_task.abort();
+
                 Ok(ExitCode::SUCCESS)
             }
         }
     }
 }
+
+/// Logs progress every 30 seconds, as a lightweight alternative to a progress
+/// bar. For most deployments, the migration will not take 30 seconds so this
+/// will not be relevant. In other cases, this will give the operator an idea of
+/// what's going on.
+async fn occasional_progress_logger(progress: Progress) {
+    loop {
+        tokio::time::sleep(Duration::from_secs(30)).await;
+        match &**progress.get_current_stage() {
+            ProgressStage::SettingUp => {
+                info!(name: "progress", "still setting up");
+            }
+            ProgressStage::MigratingData {
+                entity,
+                migrated,
+                approx_count,
+            } => {
+                let migrated = migrated.load(Ordering::Relaxed);
+                #[allow(clippy::cast_precision_loss)]
+                let percent = (f64::from(migrated) / *approx_count as f64) * 100.0;
+                info!(name: "progress", "migrating {entity}: {migrated}/~{approx_count} (~{percent:.1}%)");
+            }
+            ProgressStage::RebuildIndex { index_name } => {
+                info!(name: "progress", "still waiting for rebuild of index {index_name}");
+            }
+            ProgressStage::RebuildConstraint { constraint_name } => {
+                info!(name: "progress", "still waiting for rebuild of constraint {constraint_name}");
+            }
+        }
+    }
+}

crates/cli/src/commands/syn2mas.rs

@@ -11,6 +11,7 @@ repository.workspace = true
 
 [dependencies]
 anyhow.workspace = true
+arc-swap.workspace = true
 bitflags.workspace = true
 camino.workspace = true
 figment.workspace = true
@@ -34,6 +35,9 @@ ulid = { workspace = true, features = ["uuid"] }
 mas-config.workspace = true
 mas-storage.workspace = true
 
+opentelemetry.workspace = true
+opentelemetry-semantic-conventions.workspace = true
+
 [dev-dependencies]
 mas-storage-pg.workspace = true
 

crates/syn2mas/Cargo.toml

@@ -7,13 +7,16 @@ mod mas_writer;
 mod synapse_reader;
 
 mod migration;
+mod progress;
+mod telemetry;
 
 type RandomState = rustc_hash::FxBuildHasher;
 type HashMap<K, V> = rustc_hash::FxHashMap<K, V>;
 
 pub use self::{
     mas_writer::{MasWriter, checks::mas_pre_migration_checks, locking::LockedMasDatabase},
     migration::migrate,
+    progress::{Progress, ProgressStage},
     synapse_reader::{
         SynapseReader,
         checks::{