Revert D99820098: Delete dewey_res.bzl

Differential Revision:
D99820098

Original commit changeset: bf58ea7fb287

Original Phabricator Diff: D99820098

fbshipit-source-id: d78dc397b47ffa94cd7118932371cabbd2e80719

Author: Clara Rull

eden/mononoke/features/async_requests/worker/src/main.rs

@@ -122,8 +122,7 @@ impl RepoShardedProcess for WorkerProcess {
             .add_repo(repo_name)
             .await
             .with_context(|| format!("Failure in setting up repo {}", repo_name))?;
-        let repo_id = repo.repo_identity.id();
-        let repos = vec![repo_id];
+        let repos = vec![repo.repo_identity.id()];
         info!("Completed setup for repos {:?}", repos);
 
         let queue = Arc::new(AsyncMethodRequestQueue::new(
@@ -139,7 +138,6 @@ impl RepoShardedProcess for WorkerProcess {
             self.mononoke.clone(),
             self.megarepo.clone(),
             self.will_exit.clone(),
-            Some(repo_id),
         )
         .await?;
         Ok(Arc::new(executor))
@@ -313,7 +311,6 @@ fn run_worker_queue(
             mononoke.clone(),
             megarepo.clone(),
             will_exit.clone(),
-            None, // non-sharded: no specific repo_id
         ))?
     };
     runtime.spawn(async move { executor.execute().await });

eden/mononoke/features/async_requests/worker/src/stats.rs

@@ -5,13 +5,15 @@
  * GNU General Public License version 2.
  */
 
+use std::collections::HashMap;
 use std::time::Duration;
 
 use async_requests::AsyncMethodRequestQueue;
 use async_requests::types::RequestStatus;
 use context::CoreContext;
 use mononoke_api::RepositoryId;
 use mononoke_types::Timestamp;
+use requests_table::QueueStatsEntry;
 use stats::define_stats;
 use stats::prelude::*;
 use tracing::warn;
@@ -35,19 +37,9 @@ define_stats! {
     queue_age_by_repo_and_status: dynamic_singleton_counter("queue.{}.age_s.{}", (repo_id: String, status: String)),
 }
 
-/// Report queue stats for this shard's repo only.
-///
-/// Each shard reports metrics only for its own repo_id. This avoids
-/// cross-shard racing on shared counters and prevents stale values
-/// when a shard is reassigned: the counter simply goes absent (no
-/// data) rather than being stuck at a stale non-zero value. A dead
-/// shard detector should alert on absence of data.
-///
-/// For the non-sharded code path, `repo_id` is `None` and only the
-/// global (non-per-repo) counters are reported.
 pub(crate) async fn stats_loop(
     ctx: &CoreContext,
-    repo_id: Option<RepositoryId>,
+    repo_ids: Vec<RepositoryId>,
     queue: &AsyncMethodRequestQueue,
 ) {
     loop {
@@ -57,10 +49,8 @@ pub(crate) async fn stats_loop(
             Ok(res) => {
                 process_queue_length_by_status(ctx, &res);
                 process_queue_age_by_status(ctx, now, &res);
-                if let Some(repo_id) = &repo_id {
-                    process_queue_length_by_repo(ctx, repo_id, &res);
-                    process_queue_age_by_repo(ctx, repo_id, now, &res);
-                }
+                process_queue_length_by_repo_and_status(ctx, &repo_ids, &res);
+                process_queue_age_by_repo_and_status(ctx, &repo_ids, now, &res);
             }
             Err(err) => {
                 STATS::stats_error.add_value(1);
@@ -72,22 +62,51 @@ pub(crate) async fn stats_loop(
     }
 }
 
+// Keep track of the stats we have already logged. Any missing ones, we will log a 0.
+struct Seen {
+    inner: HashMap<QueueStatsEntry, bool>,
+}
+
+impl Seen {
+    fn new(repo_ids: &Vec<RepositoryId>) -> Self {
+        let mut seen = HashMap::new();
+        for status in STATUSES {
+            for repo_id in repo_ids {
+                seen.insert(
+                    QueueStatsEntry {
+                        repo_id: Some(repo_id.clone()),
+                        status,
+                    },
+                    false,
+                );
+            }
+        }
+        Self { inner: seen }
+    }
+
+    fn mark(&mut self, repo_id: Option<RepositoryId>, status: RequestStatus) {
+        self.inner.insert(QueueStatsEntry { repo_id, status }, true);
+    }
+
+    fn get_missing(&self) -> Vec<QueueStatsEntry> {
+        self.inner
+            .iter()
+            .filter_map(|(entry, seen)| if !*seen { Some(entry) } else { None })
+            .cloned()
+            .collect()
+    }
+}
+
 fn process_queue_length_by_status(ctx: &CoreContext, res: &requests_table::QueueStats) {
-    // Report whatever the DB returns for global stats, then zero out
-    // any of the 4 statuses that were absent from the result.
-    let mut seen = [false; 4];
+    let mut seen = Seen::new(&vec![]);
     let stats = &res.queue_length_by_status;
     for (status, count) in stats {
-        if let Some(idx) = status_index(status) {
-            seen[idx] = true;
-        }
+        seen.mark(None, *status);
         STATS::queue_length_by_status.set_value(ctx.fb, *count as i64, (status.to_string(),));
     }
 
-    for (idx, was_seen) in seen.iter().enumerate() {
-        if !was_seen {
-            STATS::queue_length_by_status.set_value(ctx.fb, 0, (STATUSES[idx].to_string(),));
-        }
+    for entry in seen.get_missing() {
+        STATS::queue_length_by_status.set_value(ctx.fb, 0, (entry.status.to_string(),));
     }
 }
 
@@ -96,93 +115,79 @@ fn process_queue_age_by_status(
     now: Timestamp,
     res: &requests_table::QueueStats,
 ) {
-    let mut seen = [false; 4];
+    let mut seen = Seen::new(&vec![]);
     let stats = &res.queue_age_by_status;
     for (status, ts) in stats {
-        if let Some(idx) = status_index(status) {
-            seen[idx] = true;
-        }
+        seen.mark(None, *status);
         let diff = std::cmp::max(now.timestamp_seconds() - ts.timestamp_seconds(), 0);
         STATS::queue_age_by_status.set_value(ctx.fb, diff, (status.to_string(),));
     }
 
-    for (idx, was_seen) in seen.iter().enumerate() {
-        if !was_seen {
-            STATS::queue_age_by_status.set_value(ctx.fb, 0, (STATUSES[idx].to_string(),));
-        }
+    for entry in seen.get_missing() {
+        STATS::queue_age_by_status.set_value(ctx.fb, 0, (entry.status.to_string(),));
     }
 }
 
-/// Report per-repo queue length for this shard's repo only.
-/// If the DB returns no data for a status, we report 0.
-fn process_queue_length_by_repo(
+fn process_queue_length_by_repo_and_status(
     ctx: &CoreContext,
-    repo_id: &RepositoryId,
+    repo_ids: &Vec<RepositoryId>,
     res: &requests_table::QueueStats,
 ) {
-    let mut seen = [false; 4];
-    let repo_id_str = repo_id.to_string();
+    let mut seen = Seen::new(repo_ids);
     let stats = &res.queue_length_by_repo_and_status;
     for (entry, count) in stats {
-        if entry.repo_id.as_ref() == Some(repo_id) {
-            if let Some(idx) = status_index(&entry.status) {
-                seen[idx] = true;
-            }
-            STATS::queue_length_by_repo_and_status.set_value(
-                ctx.fb,
-                *count as i64,
-                (repo_id_str.clone(), entry.status.to_string()),
-            );
-        }
+        seen.mark(entry.repo_id, entry.status);
+        STATS::queue_length_by_repo_and_status.set_value(
+            ctx.fb,
+            *count as i64,
+            (
+                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
+                entry.status.to_string(),
+            ),
+        );
     }
 
-    for (idx, was_seen) in seen.iter().enumerate() {
-        if !was_seen {
-            STATS::queue_length_by_repo_and_status.set_value(
-                ctx.fb,
-                0,
-                (repo_id_str.clone(), STATUSES[idx].to_string()),
-            );
-        }
+    for entry in seen.get_missing() {
+        STATS::queue_length_by_repo_and_status.set_value(
+            ctx.fb,
+            0,
+            (
+                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
+                entry.status.to_string(),
+            ),
+        );
     }
 }
 
-/// Report per-repo queue age for this shard's repo only.
-/// If the DB returns no data for a status, we report 0.
-fn process_queue_age_by_repo(
+fn process_queue_age_by_repo_and_status(
     ctx: &CoreContext,
-    repo_id: &RepositoryId,
+    repo_ids: &Vec<RepositoryId>,
     now: Timestamp,
     res: &requests_table::QueueStats,
 ) {
-    let mut seen = [false; 4];
-    let repo_id_str = repo_id.to_string();
+    let mut seen = Seen::new(repo_ids);
     let stats = &res.queue_age_by_repo_and_status;
     for (entry, ts) in stats {
-        if entry.repo_id.as_ref() == Some(repo_id) {
-            if let Some(idx) = status_index(&entry.status) {
-                seen[idx] = true;
-            }
-            let diff = std::cmp::max(now.timestamp_seconds() - ts.timestamp_seconds(), 0);
-            STATS::queue_age_by_repo_and_status.set_value(
-                ctx.fb,
-                diff,
-                (repo_id_str.clone(), entry.status.to_string()),
-            );
-        }
+        seen.mark(entry.repo_id, entry.status);
+        let diff = std::cmp::max(now.timestamp_seconds() - ts.timestamp_seconds(), 0);
+        STATS::queue_age_by_repo_and_status.set_value(
+            ctx.fb,
+            diff,
+            (
+                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
+                entry.status.to_string(),
+            ),
+        );
     }
 
-    for (idx, was_seen) in seen.iter().enumerate() {
-        if !was_seen {
-            STATS::queue_age_by_repo_and_status.set_value(
-                ctx.fb,
-                0,
-                (repo_id_str.clone(), STATUSES[idx].to_string()),
-            );
-        }
+    for entry in seen.get_missing() {
+        STATS::queue_age_by_repo_and_status.set_value(
+            ctx.fb,
+            0,
+            (
+                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
+                entry.status.to_string(),
+            ),
+        );
     }
 }
-
-fn status_index(status: &RequestStatus) -> Option<usize> {
-    STATUSES.iter().position(|s| s == status)
-}

eden/mononoke/features/async_requests/worker/src/worker.rs

@@ -42,8 +42,8 @@ use futures_stats::TimedFutureExt;
 use hostname::get_hostname;
 use megarepo_api::MegarepoApi;
 use mononoke_api::Mononoke;
+use mononoke_api::MononokeRepo;
 use mononoke_api::Repo;
-use mononoke_api::RepositoryId;
 use mononoke_macros::mononoke;
 use mononoke_types::Timestamp;
 use stats::define_stats;
@@ -86,8 +86,6 @@ pub struct AsyncMethodRequestWorker {
     will_exit: Arc<AtomicBool>,
     limit: Option<usize>,
     concurrency_limit: usize,
-    /// The repo this shard is responsible for. `None` for non-sharded workers.
-    repo_id: Option<RepositoryId>,
 }
 
 impl AsyncMethodRequestWorker {
@@ -98,7 +96,6 @@ impl AsyncMethodRequestWorker {
         mononoke: Arc<Mononoke<Repo>>,
         megarepo: Arc<MegarepoApi<Repo>>,
         will_exit: Arc<AtomicBool>,
-        repo_id: Option<RepositoryId>,
     ) -> Result<Self, Error> {
         let name = {
             let tw_job_cluster = std::env::var("TW_JOB_CLUSTER");
@@ -124,7 +121,6 @@ impl AsyncMethodRequestWorker {
             will_exit,
             limit: args.request_limit,
             concurrency_limit: args.jobs,
-            repo_id,
         })
     }
 }
@@ -137,13 +133,11 @@ impl RepoShardedProcessExecutor for AsyncMethodRequestWorker {
     /// will_exit atomic bool is a flag to prevent the worker from grabbing new
     /// items from the queue and gracefully terminate.
     async fn execute(&self) -> Result<()> {
-        // Start the stats logger loop.
-        // Each shard reports metrics only for its own repo to avoid
-        // cross-shard counter races and stale values on shard reassignment.
+        // Start the stats logger loop
         let (stats, stats_abort_handle) = abortable({
             cloned!(self.ctx, self.queue);
-            let repo_id = self.repo_id;
-            async move { stats_loop(&ctx, repo_id, &queue).await }
+            let repo_ids = self.mononoke.known_repo_ids().clone();
+            async move { stats_loop(&ctx, repo_ids, &queue).await }
         });
         let _stats = mononoke::spawn_task(stats);