fixup! syn2mas: Add progress reporting to log and to opentelemetry metrics

Add metrics directly within syn2mas, no background thread

Author: reivilibre

Cargo.lock

@@ -10,18 +10,14 @@ use mas_config::{
 };
 use mas_storage::SystemClock;
 use mas_storage_pg::MIGRATOR;
-use opentelemetry::KeyValue;
 use rand::thread_rng;
 use sqlx::{Connection, Either, PgConnection, postgres::PgConnectOptions, types::Uuid};
 use syn2mas::{
     LockedMasDatabase, MasWriter, Progress, ProgressStage, SynapseReader, synapse_config,
 };
 use tracing::{Instrument, error, info, info_span, warn};
 
-use crate::{
-    telemetry::METER,
-    util::{DatabaseConnectOptions, database_connection_from_config_with_options},
-};
+use crate::util::{DatabaseConnectOptions, database_connection_from_config_with_options};
 
 /// The exit code used by `syn2mas check` and `syn2mas migrate` when there are
 /// errors preventing migration.
@@ -258,7 +254,6 @@ impl Options {
 
                 let occasional_progress_logger_task =
                     tokio::spawn(occasional_progress_logger(progress.clone()));
-                let progress_telemetry_task = tokio::spawn(progress_telemetry(progress.clone()));
 
                 let mas_matrix = MatrixConfig::extract(figment)?;
                 eprintln!("\n\n");
@@ -274,7 +269,6 @@ impl Options {
                 .await?;
 
                 occasional_progress_logger_task.abort();
-                progress_telemetry_task.abort();
 
                 Ok(ExitCode::SUCCESS)
             }
@@ -312,32 +306,3 @@ async fn occasional_progress_logger(progress: Progress) {
         }
     }
 }
-
-/// Reports migration progress as OpenTelemetry metrics
-async fn progress_telemetry(progress: Progress) {
-    let migrated_data_counter = METER
-        .u64_gauge("migrated_data")
-        .with_description("How many entities have been migrated so far")
-        .build();
-    let max_data_counter = METER
-        .u64_gauge("max_data")
-        .with_description("How many entities of the given type exist (approximate)")
-        .build();
-
-    loop {
-        tokio::time::sleep(Duration::from_secs(10)).await;
-        if let ProgressStage::MigratingData {
-            entity,
-            migrated,
-            approx_count,
-        } = &**progress.get_current_stage()
-        {
-            let metrics_kv = [KeyValue::new("entity", *entity)];
-            let migrated = migrated.load(Ordering::Relaxed);
-            migrated_data_counter.record(u64::from(migrated), &metrics_kv);
-            max_data_counter.record(*approx_count, &metrics_kv);
-        } else {
-            // not sure how to map other stages
-        }
-    }
-}

crates/cli/src/commands/syn2mas.rs

@@ -35,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

@@ -8,6 +8,7 @@ mod synapse_reader;
 
 mod migration;
 mod progress;
+mod telemetry;
 
 type RandomState = rustc_hash::FxBuildHasher;
 type HashMap<K, V> = rustc_hash::FxHashMap<K, V>;