mononoke/fastreplay: add integration tests

Summary:
This adds integration tests for fastreplay (adapted from our existing traffic
replay test, which Stas had added a little while ago).

This works by adding support for logging to files in wireproto logging, which
we already have in a bunch of places for Scuba logging and has turned out to be
very convenient.

Since fastreplay operates directly on the JSON we generated from wireproto
logging, we can just pipe it back in, and it all works.

Reviewed By: StanislavGlebik

Differential Revision: D19577792

fbshipit-source-id: 99e6f12f5c06e7a4a69df08c0843c699b639ce54

mononoke/fastreplay/src/dispatcher.rs

@@ -6,7 +6,7 @@
  * directory of this source tree.
  */
 
-use anyhow::Error;
+use anyhow::{Context, Error};
 use blobstore::Blobstore;
 use context::{LoggingContainer, SessionContainer};
 use fbinit::FacebookInit;
@@ -32,17 +32,18 @@ impl FastReplayDispatcher {
         scuba: ScubaSampleBuilder,
         repo: MononokeRepo,
         remote_args_blobstore: Option<Arc<dyn Blobstore>>,
-    ) -> Self {
-        let noop_wireproto = WireprotoLogging::new(fb, repo.reponame().clone(), None, None);
+    ) -> Result<Self, Error> {
+        let noop_wireproto = WireprotoLogging::new(fb, repo.reponame().clone(), None, None, None)
+            .context("While instantiating noop_wireproto")?;
 
-        Self {
+        Ok(Self {
             fb,
             logger,
             scuba,
             repo,
             wireproto_logging: Arc::new(noop_wireproto),
             remote_args_blobstore,
-        }
+        })
     }
 
     pub fn client(&self) -> RepoClient {

mononoke/fastreplay/src/main.rs

@@ -238,7 +238,7 @@ async fn bootstrap_repositories<'a>(
                 scuba.clone(),
                 repo,
                 remote_args_blobstore,
-            );
+            )?;
 
             if let Some(warmup) = warmup {
                 info!(&logger, "Waiting for cache warmup to complete...");
@@ -314,7 +314,7 @@ async fn fast_replay_from_stdin<'a>(
                     continue;
                 }
                 Err(e) => {
-                    warn!(&logger, "Dispatch failed: {}", e);
+                    warn!(&logger, "Dispatch failed: {:#?}", e);
                     continue;
                 }
             },

mononoke/metaconfig/parser/src/repoconfig.rs

@@ -517,6 +517,7 @@ impl RepoConfigs {
                     scribe_category,
                     storage_config: wireproto_storage_config,
                     remote_arg_size_threshold,
+                    local_path,
                 } = wireproto_logging;
 
                 let storage_config_and_threshold = match (
@@ -543,7 +544,11 @@ impl RepoConfigs {
                     })
                     .transpose()?;
 
-                WireprotoLoggingConfig::new(scribe_category, storage_config_and_threshold)
+                WireprotoLoggingConfig {
+                    scribe_category,
+                    storage_config_and_threshold,
+                    local_path,
+                }
             }
             None => Default::default(),
         };
@@ -1584,6 +1589,7 @@ mod test {
                         main_storage_config,
                         DEFAULT_ARG_SIZE_THRESHOLD,
                     )),
+                    local_path: None,
                 },
                 hash_validation_percentage: 0,
                 readonly: RepoReadOnly::ReadWrite,

mononoke/metaconfig/types/src/lib.rs

@@ -937,19 +937,8 @@ pub struct WireprotoLoggingConfig {
     /// `storage_config_and_threshold` is not specified then wireproto wireproto arguments will
     /// be inlined
     pub storage_config_and_threshold: Option<(StorageConfig, u64)>,
-}
-
-impl WireprotoLoggingConfig {
-    /// Create WireprotoLoggingConfig with correct default values
-    pub fn new(
-        scribe_category: Option<String>,
-        storage_config_and_threshold: Option<(StorageConfig, u64)>,
-    ) -> Self {
-        Self {
-            scribe_category,
-            storage_config_and_threshold,
-        }
-    }
+    /// Local path where to log replay data that would be sent to Scribe.
+    pub local_path: Option<String>,
 }
 
 impl Default for WireprotoLoggingConfig {
@@ -957,6 +946,7 @@ impl Default for WireprotoLoggingConfig {
         Self {
             scribe_category: None,
             storage_config_and_threshold: None,
+            local_path: None,
         }
     }
 }

mononoke/repo_client/src/client/logging.rs

@@ -6,6 +6,7 @@
  * directory of this source tree.
  */
 
+use anyhow::Error;
 use blobstore::{Blobstore, BlobstoreBytes};
 use chrono::Utc;
 use context::{CoreContext, SessionId};
@@ -38,6 +39,7 @@ pub struct WireprotoLogging {
     reponame: String,
     scribe_args: Option<(ScribeClientImplementation, String)>,
     blobstore_and_threshold: Option<(Arc<dyn Blobstore>, u64)>,
+    scuba_builder: ScubaSampleBuilder,
 }
 
 impl WireprotoLogging {
@@ -46,13 +48,24 @@ impl WireprotoLogging {
         reponame: String,
         scribe_category: Option<String>,
         blobstore_and_threshold: Option<(Arc<dyn Blobstore>, u64)>,
-    ) -> Self {
+        log_file: Option<&str>,
+    ) -> Result<Self, Error> {
         let scribe_args = scribe_category.map(|cat| (ScribeClientImplementation::new(fb), cat));
-        Self {
+
+        // We use a Scuba sample builder to produce samples to log. We also use that to allow
+        // logging to a file: we never log to an actual Scuba category here.
+        let mut scuba_builder = ScubaSampleBuilder::with_discard();
+        scuba_builder.add_common_server_data();
+        if let Some(log_file) = log_file {
+            scuba_builder = scuba_builder.with_log_file(log_file)?;
+        }
+
+        Ok(Self {
             reponame,
             scribe_args,
             blobstore_and_threshold,
-        }
+            scuba_builder,
+        })
     }
 }
 
@@ -201,9 +214,8 @@ fn do_wireproto_logging<'a>(
 
     // Use a ScubaSampleBuilder to build a sample to send in Scribe. Reach into the other Scuba
     // sample to grab a few datapoints from there as well.
-    let mut builder = ScubaSampleBuilder::with_discard();
+    let mut builder = wireproto.scuba_builder.clone();
     builder
-        .add_common_server_data()
         .add("command", command)
         .add("duration", stats.completion_time().as_micros_unchecked())
         .add("source_control_server_type", "mononoke")
@@ -247,7 +259,11 @@ fn do_wireproto_logging<'a>(
         };
 
         prepare_fut
-            .map(move |builder| {
+            .map(move |mut builder| {
+                // We use the Scuba sample and log it to Scribe, then we also log in the Scuba
+                // sample, but this is built using discard(), so at most it'll log to a file for
+                // debug / tests.
+
                 let sample = builder.get_sample();
                 // We can't really do anything with the errors, so let's just log them
                 if let Some((ref scribe_client, ref scribe_category)) = wireproto.scribe_args {
@@ -262,6 +278,8 @@ fn do_wireproto_logging<'a>(
                         STATS::wireproto_serialization_failure.add_value(1);
                     }
                 }
+
+                builder.log();
             })
             .or_else(|_| Ok(()))
     });

mononoke/server/repo_listener/src/repo_handlers.rs

@@ -435,6 +435,7 @@ fn create_wireproto_logging(
     let WireprotoLoggingConfig {
         storage_config_and_threshold,
         scribe_category,
+        local_path,
     } = wireproto_logging_config;
     let blobstore_fut = match storage_config_and_threshold {
         Some((storage_config, threshold)) => {
@@ -452,8 +453,14 @@ fn create_wireproto_logging(
     };
 
     blobstore_fut
-        .map(move |blobstore_and_threshold| {
-            WireprotoLogging::new(fb, reponame, scribe_category, blobstore_and_threshold)
+        .and_then(move |blobstore_and_threshold| {
+            WireprotoLogging::new(
+                fb,
+                reponame,
+                scribe_category,
+                blobstore_and_threshold,
+                local_path.as_ref().map(|p| p.as_ref()),
+            )
         })
         .left_future()
 }

mononoke/tests/integration/library.sh

@@ -605,8 +605,14 @@ list_keys_patterns_max=$LIST_KEYS_PATTERNS_MAX
 CONFIG
 fi
 
+if [[ -v WIREPROTO_LOGGING_PATH ]]; then
   cat >> "repos/$reponame/server.toml" <<CONFIG
 [wireproto_logging]
+local_path="$WIREPROTO_LOGGING_PATH"
+CONFIG
+
+  if [[ -v WIREPROTO_LOGGING_BLOBSTORE ]]; then
+    cat >> "repos/$reponame/server.toml" <<CONFIG
 storage_config="traffic_replay_blobstore"
 remote_arg_size_threshold=0
 
@@ -614,10 +620,11 @@ remote_arg_size_threshold=0
 local_db_path="$TESTTMP/monsql"
 
 [storage.traffic_replay_blobstore.blobstore.blob_files]
-path = "$TESTTMP/traffic-replay-blobstore"
-
+path = "$WIREPROTO_LOGGING_BLOBSTORE"
 CONFIG
-
+  fi
+fi
+# path = "$TESTTMP/traffic-replay-blobstore"
 
 if [[ -v ONLY_FAST_FORWARD_BOOKMARK ]]; then
   cat >> "repos/$reponame/server.toml" <<CONFIG
@@ -1206,7 +1213,7 @@ function traffic_replay() {
     --testrun \
     --hgcli "$MONONOKE_HGCLI" \
     --mononoke-address "[::1]:$MONONOKE_SOCKET" \
-    --mononoke-server-common-name localhost < "$1"
+    --mononoke-server-common-name localhost
 }
 
 function enable_replay_verification_hook {
@@ -1432,3 +1439,23 @@ function regenerate_hg_filenodes() {
     --i-know-what-i-am-doing \
     "$@"
 }
+
+function fastreplay() {
+  "$MONONOKE_FASTREPLAY" \
+    "${COMMON_ARGS[@]}" \
+    --no-skiplist \
+    --no-cache-warmup \
+    --mononoke-config-path "${TESTTMP}/mononoke-config" \
+    "$@"
+}
+
+function quiet() {
+  local log="$TESTTMP/quiet.last.log"
+  "$@" >"$log" 2>&1
+  ret="$?"
+  if [[ "$ret" == 0 ]]; then
+    return "$ret"
+  fi
+  cat "$log"
+  return "$ret"
+}

mononoke/tests/integration/test-fastreplay-inline-args.t

@@ -0,0 +1,34 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# This software may be used and distributed according to the terms of the
+# GNU General Public License found in the LICENSE file in the root
+# directory of this source tree.
+
+  $ . "${TEST_FIXTURES}/library.sh"
+
+Setup configuration
+  $ export WIREPROTO_LOGGING_PATH="$TESTTMP/wireproto.json"
+  $ BLOB_TYPE="blob_files" quiet default_setup
+
+Make requests
+  $ quiet hgmn  pull
+  $ quiet hgmn up master_bookmark
+
+Wait for requests to be logged
+  $ wait_for_json_record_count "$WIREPROTO_LOGGING_PATH" 3
+  $ jq -r .normal.command "$WIREPROTO_LOGGING_PATH"
+  getbundle
+  gettreepack
+  getpackv1
+
+Replay traffic
+  $ fastreplay_log="$TESTTMP/fastreplay.json"
+  $ quiet fastreplay --scuba-log-file "$fastreplay_log" < "$WIREPROTO_LOGGING_PATH"
+  $ jq -r .normal.command  "$fastreplay_log" | grep -v null | sort | uniq
+  getbundle
+  getpackv1
+  gettreepack
+  $ jq -r .normal.log_tag  "$fastreplay_log" | grep Replay
+  Replay Succeeded
+  Replay Succeeded
+  Replay Succeeded

mononoke/tests/integration/test-fastreplay-remote-args.t

@@ -0,0 +1,54 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# This software may be used and distributed according to the terms of the
+# GNU General Public License found in the LICENSE file in the root
+# directory of this source tree.
+
+  $ . "${TEST_FIXTURES}/library.sh"
+
+setup configuration
+  $ export WIREPROTO_LOGGING_PATH="$TESTTMP/wireproto.json"
+  $ export WIREPROTO_LOGGING_BLOBSTORE="$TESTTMP/traffic-replay-blobstore"
+  $ BLOB_TYPE="blob_files" quiet default_setup
+
+Make requests
+  $ quiet hgmn  pull
+  $ quiet hgmn up master_bookmark
+
+Wait for requests to be logged
+  $ wait_for_json_record_count "$WIREPROTO_LOGGING_PATH" 3
+  $ jq -r .normal.command "$WIREPROTO_LOGGING_PATH"
+  getbundle
+  gettreepack
+  getpackv1
+
+Replay traffic using the ephemeral blobstore
+  $ fastreplay_log="$TESTTMP/fastreplay.json"
+  $ quiet fastreplay --scuba-log-file "$fastreplay_log" < "$WIREPROTO_LOGGING_PATH"
+  $ jq -r .normal.command  "$fastreplay_log" | grep -v null | sort | uniq
+  getbundle
+  getpackv1
+  gettreepack
+  $ jq -r .normal.log_tag  "$fastreplay_log" | grep Replay
+  Replay Succeeded
+  Replay Succeeded
+  Replay Succeeded
+
+Delete the ephemeral blobstore data. Check that replay now fails.
+  $ rm -r "$WIREPROTO_LOGGING_BLOBSTORE"
+  $ fastreplay < "$WIREPROTO_LOGGING_PATH"
+  * Creating 1 repositories (glob)
+  * Repositories are ready! (glob)
+  * Dispatch failed: Error { (glob)
+      context: "While loading remote_args",
+      source: "Key not found: wireproto_replay.*", (glob)
+  }
+  * Dispatch failed: Error { (glob)
+      context: "While loading remote_args",
+      source: "Key not found: wireproto_replay.*", (glob)
+  }
+  * Dispatch failed: Error { (glob)
+      context: "While loading remote_args",
+      source: "Key not found: wireproto_replay.*", (glob)
+  }
+  * Processed all input... (glob)

mononoke/tests/integration/test-traffic-replay.t

@@ -6,70 +6,20 @@
 
   $ . "${TEST_FIXTURES}/library.sh"
 
-setup configuration
-  $ setup_common_config "blob_files"
-  $ cd $TESTTMP
+Setup configuration
+  $ export WIREPROTO_LOGGING_PATH="$TESTTMP/wireproto.json"
+  $ BLOB_TYPE="blob_files" quiet default_setup
 
-setup common configuration
-  $ cat >> $HGRCPATH <<EOF
-  > [ui]
-  > ssh="$DUMMYSSH"
-  > [extensions]
-  > amend=
-  > EOF
+Make requests
+  $ quiet hgmn pull
 
-Setup helpers
-  $ log() {
-  >   hg log -G -T "{desc} [{phase};rev={rev};{node|short}] {remotenames}" "$@"
-  > }
+Wait for requests to be logged
+  $ wait_for_json_record_count "$WIREPROTO_LOGGING_PATH" 1
+  $ jq -r .normal.command "$WIREPROTO_LOGGING_PATH"
+  getbundle
 
-setup repo
-  $ hg init repo-hg
-  $ cd repo-hg
-  $ setup_hg_server
-  $ hg debugdrawdag <<EOF
-  > C
-  > |
-  > B
-  > |
-  > A
-  > EOF
+Replay traffic
+  $ quiet traffic_replay < "$WIREPROTO_LOGGING_PATH"
 
-create master bookmark
-
-  $ hg bookmark master_bookmark -r tip
-
-blobimport them into Mononoke storage and start Mononoke
-  $ cd ..
-  $ blobimport repo-hg/.hg repo
-
-start mononoke
-  $ mononoke
-  $ wait_for_mononoke $TESTTMP/repo
-
-Clone the repo
-  $ hgclone_treemanifest ssh://user@dummy/repo-hg repo2 --noupdate --config extensions.remotenames= -q
-  $ cd repo2
-  $ setup_hg_client
-  $ cat >> .hg/hgrc <<EOF
-  > [extensions]
-  > pushrebase =
-  > remotenames =
-  > EOF
-
-  $ hgmn pull -q
-  devel-warn: applied empty changegroup at: * (glob)
-
-We are going to put these args inside traffic replay request below.
-For that we need to escape all double quotes. `sed` below does exactly that
-  $ ARGS=$(sed 's/"/\\"/g' < "$TESTTMP/traffic-replay-blobstore/blobs/"*)
-  $ cat >> traffic_replay_request <<EOF
-  > {"int":{"duration":0}, "normal":{"command":"getbundle","args":"$ARGS", "reponame":"$REPONAME"}}
-  > EOF
-  $ ls -l $TESTTMP/traffic-replay-blobstore/blobs/ | grep blob | wc -l
-  1
-  $ traffic_replay traffic_replay_request &> /dev/null
-
-Make sure one more was added to the ephemeral blobstore
-  $ ls -l $TESTTMP/traffic-replay-blobstore/blobs/ | grep blob | wc -l
-  2
+Make sure more traffic was logged
+  $ wait_for_json_record_count "$WIREPROTO_LOGGING_PATH" 2