modern sync: fix stats

Summary: With the new syncing design we're logging when we queue the changeset, not when we actually send it and succeed. Fixing that by adding another message through the changesets channel so when  the processing of the changeste is done we do the logging.

Reviewed By: andreacampi

Differential Revision: D68926376

fbshipit-source-id: 3a17b7567d822616b4e149b963464b1ad9103f3b

eden/mononoke/modern_sync/src/sender/manager.rs

@@ -17,10 +17,19 @@ use mononoke_types::BonsaiChangeset;
 use mononoke_types::FileContents;
 use slog::error;
 use slog::Logger;
+use stats::define_stats;
+use stats::prelude::*;
 use tokio::sync::mpsc;
 
 use crate::sender::ModernSyncSender;
 
+define_stats! {
+    prefix = "mononoke.modern_sync";
+    completion_duration_secs: timeseries(Average, Sum, Count),
+    synced_commits:  dynamic_timeseries("{}.commits_synced", (repo: String); Rate, Sum),
+    sync_lag_seconds:  dynamic_timeseries("{}.sync_lag_seconds", (repo: String); Average),
+}
+
 const CONTENT_CHANNEL_SIZE: usize = 1000;
 const FILES_AND_TREES_CHANNEL_SIZE: usize = 1000;
 const CHANGESET_CHANNEL_SIZE: usize = 1000;
@@ -57,6 +66,8 @@ pub enum ChangesetMessage {
     Changeset((HgBlobChangeset, BonsaiChangeset)),
     // Notify changeset sending is done
     ChangesetDone(mpsc::Sender<Result<()>>),
+    // Log changeset completion
+    Log((String, Option<i64>)),
 }
 
 impl SendManager {
@@ -232,6 +243,12 @@ impl SendManager {
                             }
                         }
                     }
+                    ChangesetMessage::Log((reponame, lag)) => {
+                        STATS::synced_commits.add_value(1, (reponame.clone(),));
+                        if let Some(lag) = lag {
+                            STATS::sync_lag_seconds.add_value(lag, (reponame,));
+                        }
+                    }
                 }
             }
         });

eden/mononoke/modern_sync/src/sync.rs

@@ -51,7 +51,6 @@ use repo_derived_data::RepoDerivedDataRef;
 use repo_identity::RepoIdentityRef;
 use slog::info;
 use slog::Logger;
-use stats::prelude::*;
 use tokio::sync::mpsc;
 use url::Url;
 
@@ -67,13 +66,6 @@ use crate::ModernSyncArgs;
 use crate::Repo;
 const MODERN_SYNC_COUNTER_NAME: &str = "modern_sync";
 
-define_stats! {
-    prefix = "mononoke.modern_sync";
-    completion_duration_secs: timeseries(Average, Sum, Count),
-    synced_commits:  dynamic_timeseries("{}.commits_synced", (repo: String); Rate, Sum),
-    sync_lag_seconds:  dynamic_timeseries("{}.sync_lag_seconds", (repo: String); Average),
-}
-
 #[derive(Clone)]
 pub enum ExecutionType {
     SyncOnce,
@@ -435,23 +427,26 @@ pub async fn process_one_changeset(
     }
 
     if log_to_ods {
-        STATS::synced_commits.add_value(1, (repo.repo_identity().name().to_string(),));
-
-        if let Some(cs_id) = repo
+        let lag = if let Some(cs_id) = repo
             .bookmarks()
             .get(ctx.clone(), &BookmarkKey::new(bookmark_name)?)
             .await?
         {
             let bookmark_commit = cs_id.load(ctx, repo.repo_blobstore()).await?;
             let bookmark_commit_time = bookmark_commit.author_date().timestamp_secs();
 
-            STATS::sync_lag_seconds.add_value(
-                bookmark_commit_time - commit_time,
-                (repo.repo_identity().name().to_string(),),
-            );
+            Some(bookmark_commit_time - commit_time)
         } else {
             info!(logger, "Bookmark {} not found", bookmark_name);
-        }
+            None
+        };
+
+        send_manager
+            .send_changeset(ChangesetMessage::Log((
+                repo.repo_identity().name().to_string(),
+                lag,
+            )))
+            .await?;
     }
 
     Ok(())