Command to manually scrub keys supplied on stdin

Summary: For populating the XDB blobstore, we'd like to copy data from Manifold - the easiest way to do that is to exploit MultiplexedBlobstore's scrub mode to copy data directly.

Reviewed By: krallin

Differential Revision: D22373838

fbshipit-source-id: 550a9c73e79059380337fa35ac94fe1134378196

eden/mononoke/Cargo.toml

@@ -4,7 +4,7 @@ edition = "2018"
 version = "0.1.0"
 authors = ['Facebook']
 license = "GPLv2+"
-include = ["cmds/aliasverify.rs", "cmds/backfill_derived_data/**/*.rs", "cmds/backfill_git_mapping.rs", "cmds/benchmark_storage_config/**/*.rs", "cmds/bonsai_verify/**/*.rs", "cmds/configlint.rs", "cmds/dumprev.rs", "cmds/idxdump.rs", "cmds/lfs_import.rs", "cmds/rechunker.rs", "cmds/revlogrepo.rs", "cmds/statistics_collector.rs", "cmds/upload_globalrevs.rs"]
+include = ["cmds/aliasverify.rs", "cmds/backfill_derived_data/**/*.rs", "cmds/backfill_git_mapping.rs", "cmds/benchmark_storage_config/**/*.rs", "cmds/bonsai_verify/**/*.rs", "cmds/configlint.rs", "cmds/dumprev.rs", "cmds/idxdump.rs", "cmds/lfs_import.rs", "cmds/manual_scrub/**/*.rs", "cmds/rechunker.rs", "cmds/revlogrepo.rs", "cmds/statistics_collector.rs", "cmds/upload_globalrevs.rs"]
 
 [[bin]]
 name = "aliasverify"
@@ -42,6 +42,10 @@ path = "cmds/idxdump.rs"
 name = "lfs_import"
 path = "cmds/lfs_import.rs"
 
+[[bin]]
+name = "manual_scrub"
+path = "cmds/manual_scrub/main.rs"
+
 [[bin]]
 name = "rechunker"
 path = "cmds/rechunker.rs"
@@ -88,6 +92,7 @@ metaconfig_types = { path = "metaconfig/types" }
 mononoke_types = { path = "mononoke_types" }
 revset = { path = "revset" }
 scuba_ext = { path = "common/scuba_ext" }
+sql_ext = { path = "common/rust/sql_ext" }
 unodes = { path = "derived_data/unodes" }
 cloned = { git = "https://github.com/facebookexperimental/rust-shed.git", branch = "master" }
 failure_ext = { git = "https://github.com/facebookexperimental/rust-shed.git", branch = "master" }

eden/mononoke/cmds/manual_scrub/blobstore.rs

@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This software may be used and distributed according to the terms of the
+ * GNU General Public License version 2.
+ */
+
+use std::sync::Arc;
+
+use anyhow::{Error, Result};
+use fbinit::FacebookInit;
+use slog::Logger;
+
+use blobstore::Blobstore;
+use blobstore_factory::{make_blobstore, BlobstoreOptions};
+use metaconfig_types::{ScrubAction, StorageConfig};
+use sql_ext::facebook::MysqlOptions;
+
+pub async fn open_blobstore(
+    fb: FacebookInit,
+    mut storage_config: StorageConfig,
+    mysql_options: MysqlOptions,
+    blobstore_options: &BlobstoreOptions,
+    logger: &Logger,
+) -> Result<Arc<dyn Blobstore>> {
+    storage_config.blobstore.set_scrubbed(ScrubAction::Repair);
+
+    make_blobstore(
+        fb,
+        storage_config.blobstore,
+        mysql_options,
+        blobstore_factory::ReadOnlyStorage(false),
+        blobstore_options,
+        logger,
+    )
+    .await
+    .map_err(Error::from)
+}

eden/mononoke/cmds/manual_scrub/main.rs

@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This software may be used and distributed according to the terms of the
+ * GNU General Public License version 2.
+ */
+
+#![deny(warnings)]
+
+use anyhow::{Context, Error, Result};
+use clap::Arg;
+use futures::{
+    channel::mpsc,
+    stream::{StreamExt, TryStreamExt},
+};
+use tokio::io::{stdin, stdout, AsyncBufReadExt, AsyncWriteExt, BufReader};
+
+use cmdlib::args;
+use context::CoreContext;
+
+mod blobstore;
+mod scrub;
+
+use crate::{blobstore::open_blobstore, scrub::scrub};
+
+const ARG_STORAGE_CONFIG_NAME: &str = "storage-config-name";
+
+async fn bridge_to_stdout(mut recv: mpsc::Receiver<String>) -> Result<()> {
+    let mut stdout = stdout();
+    while let Some(string) = recv.next().await {
+        stdout.write_all(string.as_bytes()).await?;
+        stdout.write(b"\n").await?;
+    }
+    Ok(())
+}
+
+#[fbinit::main]
+fn main(fb: fbinit::FacebookInit) -> Result<()> {
+    let app = args::MononokeApp::new("manual scrub")
+        .with_advanced_args_hidden()
+        .with_all_repos()
+        .build()
+        .arg(
+            Arg::with_name(ARG_STORAGE_CONFIG_NAME)
+                .long(ARG_STORAGE_CONFIG_NAME)
+                .takes_value(true)
+                .required(true)
+                .help("the name of the storage config to scrub"),
+        );
+    let matches = app.get_matches();
+    let (_, logger, mut runtime) =
+        args::init_mononoke(fb, &matches, None).context("failed to initialise mononoke")?;
+
+    let storage_config = args::load_storage_configs(fb, &matches)
+        .context("Could not read storage configs")?
+        .storage
+        .remove(
+            matches
+                .value_of(ARG_STORAGE_CONFIG_NAME)
+                .context("No storage config name")?,
+        )
+        .context("Requested storage config not found")?;
+
+    let mysql_options = args::parse_mysql_options(&matches);
+    let blobstore_options = args::parse_blobstore_options(&matches);
+    let ctx = CoreContext::new_with_logger(fb, logger.clone());
+
+    let scrub = async move {
+        let blobstore = open_blobstore(
+            fb,
+            storage_config,
+            mysql_options,
+            &blobstore_options,
+            &logger,
+        )
+        .await?;
+
+        let stdin = BufReader::new(stdin());
+        let (output, recv) = mpsc::channel(100);
+        let output_handle = tokio::spawn(bridge_to_stdout(recv));
+        let res = scrub(
+            &*blobstore,
+            &ctx,
+            stdin.lines().map_err(Error::from),
+            output,
+        )
+        .await
+        .context("Scrub failed");
+
+        output_handle.await?.context("Writing to stdout failed")?;
+        res
+    };
+
+    runtime.block_on_std(scrub)
+}

eden/mononoke/cmds/manual_scrub/scrub.rs

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This software may be used and distributed according to the terms of the
+ * GNU General Public License version 2.
+ */
+
+use anyhow::{Context, Result};
+use futures::{
+    channel::mpsc,
+    sink::SinkExt,
+    stream::{Stream, TryStreamExt},
+};
+
+use blobstore::Blobstore;
+use context::CoreContext;
+
+async fn scrub_key(
+    blobstore: &dyn Blobstore,
+    ctx: &CoreContext,
+    key: String,
+    mut output: mpsc::Sender<String>,
+) -> Result<()> {
+    blobstore
+        .get(ctx.clone(), key.clone())
+        .await?
+        .with_context(|| format!("Key {} is missing", &key))?;
+    output.send(key).await?;
+    Ok(())
+}
+
+pub async fn scrub(
+    blobstore: &dyn Blobstore,
+    ctx: &CoreContext,
+    keys: impl Stream<Item = Result<String>>,
+    output: mpsc::Sender<String>,
+) -> Result<()> {
+    keys.try_for_each_concurrent(100, |key| scrub_key(blobstore, ctx, key, output.clone()))
+        .await?;
+    Ok(())
+}

eden/mononoke/tests/integration/library.sh

@@ -906,6 +906,11 @@ function lfs_import {
   --mononoke-config-path "$TESTTMP/mononoke-config" "${COMMON_ARGS[@]}" "$@"
 }
 
+function manual_scrub {
+  GLOG_minloglevel=5 "$MONONOKE_MANUAL_SCRUB" \
+  --mononoke-config-path "$TESTTMP/mononoke-config" "${COMMON_ARGS[@]}" "$@"
+}
+
 function s_client {
     /usr/local/fbcode/platform007/bin/openssl s_client \
         -connect localhost:$MONONOKE_SOCKET \

eden/mononoke/tests/integration/test-cmd-manual-scrub.t

@@ -0,0 +1,58 @@
+# 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.
+
+setup
+  $ . "${TEST_FIXTURES}/library.sh"
+
+setup configuration
+  $ MULTIPLEXED=2 default_setup_blobimport "blob_files"
+  hg repo
+  o  C [draft;rev=2;26805aba1e60]
+  |
+  o  B [draft;rev=1;112478962961]
+  |
+  o  A [draft;rev=0;426bada5c675]
+  $
+  blobimporting
+
+Run a heal
+  $ mononoke_blobstore_healer -q --iteration-limit=1 --heal-min-age-secs=0 --storage-id=blobstore --sync-queue-limit=100 2>&1 > /dev/null
+
+Failure time - this key will not exist
+  $ manual_scrub --storage-config-name blobstore <<EOF
+  > fake-key
+  > EOF
+  Error: Scrub failed
+  
+  Caused by:
+      Key fake-key is missing
+  [1]
+
+Success time - these keys will exist and be scrubbed
+  $ manual_scrub --storage-config-name blobstore <<EOF
+  > repo0000.hgchangeset.sha1.26805aba1e600a82e93661149f2313866a221a7b
+  > repo0000.content.blake2.55662471e2a28db8257939b2f9a2d24e65b46a758bac12914a58f17dcde6905f
+  > repo0000.hgfilenode.sha1.35e7525ce3a48913275d7061dd9a867ffef1e34d
+  > EOF
+  repo0000.hgchangeset.sha1.26805aba1e600a82e93661149f2313866a221a7b
+  repo0000.content.blake2.55662471e2a28db8257939b2f9a2d24e65b46a758bac12914a58f17dcde6905f
+  repo0000.hgfilenode.sha1.35e7525ce3a48913275d7061dd9a867ffef1e34d
+
+Demostrate that a key exists
+  $ ls "$TESTTMP/blobstore/0/blobs/blob-repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0"
+  $TESTTMP/blobstore/0/blobs/blob-repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0
+Delete it
+  $ rm "$TESTTMP/blobstore/0/blobs/blob-repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0"
+Scrub restores it
+  $ manual_scrub --storage-config-name blobstore <<EOF
+  > repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0
+  > EOF
+  * scrub: blobstore_id BlobstoreId(0) repaired for repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0 (glob)
+  repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0
+Run a heal and demonstrate that it's back
+  $ mononoke_blobstore_healer -q --iteration-limit=1 --heal-min-age-secs=0 --storage-id=blobstore --sync-queue-limit=100 2>&1 > /dev/null
+  $ ls "$TESTTMP/blobstore/0/blobs/blob-repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0"
+  $TESTTMP/blobstore/0/blobs/blob-repo0000.hgchangeset.sha1.426bada5c67598ca65036d57d9e4b64b0c1ce7a0