multiplex_wal: record the multiplex member that served each get request

Summary: The multiplexed blobstore doesn't record which of the multiplex members won the race to serve the request.  Let's change that so we can analyze blobstore usage.

Reviewed By: quark-zju

Differential Revision: D75457108

fbshipit-source-id: 543e324900e00ec00a1bee14e08b4e0166ad85cf

Author: markbt

eden/mononoke/blobstore/multiplexedblob/src/scuba.rs

@@ -28,6 +28,7 @@ const MULTIPLEX_ID: &str = "multiplex_id";
 const BLOB_SIZE: &str = "blob_size";
 const SUCCESS: &str = "success";
 const SYNC_QUEUE: &str = "mysql_sync_queue";
+const BLOBSTORE_ID: &str = "blobstore_id";
 
 fn record_scuba_common(
     mut ctx: CoreContext,
@@ -102,7 +103,7 @@ pub fn record_get(
     multiplex_id: &MultiplexId,
     key: &str,
     stats: FutureStats,
-    result: &Result<Option<BlobstoreGetData>>,
+    result: &Result<Option<(BlobstoreId, BlobstoreGetData)>>,
 ) {
     let op = OperationType::Get;
     record_scuba_common(ctx.clone(), scuba, multiplex_id, key, stats, op);
@@ -116,7 +117,8 @@ pub fn record_get(
             let blob_present = mb_blob.is_some();
             scuba.add(BLOB_PRESENT, blob_present).add(SUCCESS, true);
 
-            if let Some(blob) = mb_blob.as_ref() {
+            if let Some((bs_id, blob)) = mb_blob.as_ref() {
+                scuba.add(BLOBSTORE_ID, *bs_id);
                 scuba.add(BLOB_SIZE, blob.len());
             }
         }

eden/mononoke/blobstore/multiplexedblob_wal/src/multiplex.rs

@@ -445,7 +445,7 @@ impl WalMultiplexedBlobstore {
         ctx: &'a CoreContext,
         key: &'a str,
         scuba: &Scuba,
-    ) -> Result<Option<BlobstoreGetData>> {
+    ) -> Result<Option<(BlobstoreId, BlobstoreGetData)>> {
         ctx.perf_counters()
             .increment_counter(PerfCounterType::BlobGets);
 
@@ -466,7 +466,7 @@ impl WalMultiplexedBlobstore {
             while let Some((bs_id, result)) = get_futs.next().await {
                 match result {
                     Ok(Some(get_data)) => {
-                        return Ok(Some(get_data));
+                        return Ok(Some((bs_id, get_data)));
                     }
                     Ok(None) => {
                         quorum = quorum.saturating_sub(1);
@@ -504,7 +504,7 @@ impl WalMultiplexedBlobstore {
         });
 
         match result {
-            Ok(Some(ref data)) => {
+            Ok(Some((_bs_id, ref data))) => {
                 ctx.perf_counters()
                     .set_max_counter(PerfCounterType::BlobGetsMaxSize, data.len() as i64);
                 ctx.perf_counters()
@@ -624,7 +624,7 @@ impl Blobstore for WalMultiplexedBlobstore {
             stats,
             &result,
         );
-        result
+        Ok(result?.map(|(_, data)| data))
     }
 
     async fn is_present<'a>(