br: support checkSum during stream restore dml kv-events

components/backup-stream/src/metadata/client.rs

@@ -13,7 +13,6 @@ use kvproto::brpb::StreamBackupTaskInfo;
 use tikv_util::{defer, time::Instant, warn};
 use tokio_stream::StreamExt;
 
-
 use crate::errors::{Error, Result};
 
 /// Some operations over stream backup metadata key space.

components/backup-stream/src/observer.rs

@@ -243,7 +243,7 @@ mod tests {
     use raft::StateRole;
     use raftstore::coprocessor::{
         Cmd, CmdBatch, CmdObserveInfo, CmdObserver, ObserveHandle, ObserveLevel, ObserverContext,
-        RegionChangeEvent, RegionChangeObserver, RoleObserver, RoleChange,
+        RegionChangeEvent, RegionChangeObserver, RoleChange, RoleObserver,
     };
 
     use tikv_util::worker::dummy_scheduler;

components/backup-stream/src/router.rs

@@ -1019,7 +1019,7 @@ impl DataFile {
         self.set_storage_path(file_key.file_name(self.min_ts, self.max_ts));
 
         let mut meta = DataFileInfo::new();
-        meta.set_sha_256(
+        meta.set_sha256(
             self.sha256
                 .finish()
                 .map(|bytes| bytes.to_vec())

components/external_storage/Cargo.toml

@@ -43,6 +43,7 @@ tokio = { version = "1.5", features = ["time", "fs", "process"] }
 tokio-util = { version = "0.6", features = ["compat"] }
 url = "2.0"
 async-trait = "0.1"
+openssl = "0.10"
 
 [dev-dependencies]
 structopt = "0.3"

components/external_storage/export/src/export.rs

@@ -325,6 +325,7 @@ impl ExternalStorage for EncryptedExternalStorage {
         storage_name: &str,
         restore_name: std::path::PathBuf,
         expected_length: u64,
+        expected_sha256: Option<Vec<u8>>,
         speed_limiter: &Limiter,
         file_crypter: Option<FileEncryptionInfo>,
     ) -> io::Result<()> {
@@ -339,6 +340,7 @@ impl ExternalStorage for EncryptedExternalStorage {
             file_writer,
             speed_limiter,
             expected_length,
+            expected_sha256,
             min_read_speed,
         ))
     }

components/external_storage/export/src/request.rs

@@ -69,6 +69,7 @@ pub async fn restore_inner(
         output,
         &limiter,
         expected_length,
+        None,
         MINIMUM_READ_SPEED,
     )
     .await;

components/external_storage/src/lib.rs

@@ -20,6 +20,7 @@ use engine_traits::FileEncryptionInfo;
 use file_system::File;
 use futures_io::AsyncRead;
 use futures_util::AsyncReadExt;
+use openssl::hash::{Hasher, MessageDigest};
 use tikv_util::stream::{block_on_external_io, READ_BUF_SIZE};
 use tikv_util::time::{Instant, Limiter};
 use tokio::time::timeout;
@@ -78,6 +79,7 @@ pub trait ExternalStorage: 'static + Send + Sync {
         storage_name: &str,
         restore_name: std::path::PathBuf,
         expected_length: u64,
+        expected_sha256: Option<Vec<u8>>,
         speed_limiter: &Limiter,
         file_crypter: Option<FileEncryptionInfo>,
     ) -> io::Result<()> {
@@ -100,6 +102,7 @@ pub trait ExternalStorage: 'static + Send + Sync {
             output,
             speed_limiter,
             expected_length,
+            expected_sha256,
             min_read_speed,
         ))
     }
@@ -143,8 +146,8 @@ impl ExternalStorage for Box<dyn ExternalStorage> {
     }
 }
 
-// Wrap the reader with file_crypter
-// Return the reader directly if file_crypter is None
+/// Wrap the reader with file_crypter.
+/// Return the reader directly if file_crypter is None.
 pub fn encrypt_wrap_reader<'a>(
     file_crypter: Option<FileEncryptionInfo>,
     reader: Box<dyn AsyncRead + Unpin + 'a>,
@@ -167,13 +170,20 @@ pub async fn read_external_storage_into_file(
     output: &mut dyn Write,
     speed_limiter: &Limiter,
     expected_length: u64,
+    expected_sha256: Option<Vec<u8>>,
     min_read_speed: usize,
 ) -> io::Result<()> {
     let dur = Duration::from_secs((READ_BUF_SIZE / min_read_speed) as u64);
 
     // do the I/O copy from external_storage to the local file.
     let mut buffer = vec![0u8; READ_BUF_SIZE];
     let mut file_length = 0;
+    let mut hasher = Hasher::new(MessageDigest::sha256()).map_err(|err| {
+        io::Error::new(
+            io::ErrorKind::Other,
+            format!("openssl hasher failed to init: {}", err),
+        )
+    })?;
 
     loop {
         // separate the speed limiting from actual reading so it won't
@@ -186,6 +196,14 @@ pub async fn read_external_storage_into_file(
         }
         speed_limiter.consume(bytes_read).await;
         output.write_all(&buffer[..bytes_read])?;
+        if expected_sha256.is_some() {
+            hasher.update(&buffer[..bytes_read]).map_err(|err| {
+                io::Error::new(
+                    io::ErrorKind::Other,
+                    format!("openssl hasher udpate failed: {}", err),
+                )
+            })?;
+        }
         file_length += bytes_read as u64;
     }
 
@@ -199,5 +217,26 @@ pub async fn read_external_storage_into_file(
         ));
     }
 
+    if let Some(expected_s) = expected_sha256 {
+        let cal_sha256 = hasher.finish().map_or_else(
+            |err| {
+                Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!("openssl hasher finish failed: {}", err),
+                ))
+            },
+            |bytes| Ok(bytes.to_vec()),
+        )?;
+        if !expected_s.eq(&cal_sha256) {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "sha256 not match, expect: {:?}, calculate: {:?}",
+                    expected_s, cal_sha256,
+                ),
+            ));
+        }
+    }
+
     Ok(())
 }

components/pd_client/src/lib.rs

@@ -21,8 +21,8 @@ pub use self::util::{merge_bucket_stats, new_bucket_stats};
 use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::ops::Deref;
-use std::time::Duration;
 use std::sync::Arc;
+use std::time::Duration;
 
 use futures::future::BoxFuture;
 use grpcio::ClientSStreamReceiver;

components/sst_importer/Cargo.toml

@@ -39,6 +39,7 @@ tikv_util = { path = "../tikv_util", default-features = false }
 tokio = { version = "1.5", features = ["time", "rt-multi-thread", "macros"] }
 txn_types = { path = "../txn_types", default-features = false }
 uuid = { version = "0.8.1", features = ["serde", "v4"] }
+openssl = "0.10"
 
 [dev-dependencies]
 tempfile = "3.0"

components/sst_importer/src/sst_importer.rs

@@ -212,6 +212,7 @@ impl SSTImporter {
         src_file_name: &str,
         dst_file: std::path::PathBuf,
         backend: &StorageBackend,
+        expect_sha256: Option<Vec<u8>>,
         file_crypter: Option<FileEncryptionInfo>,
         speed_limiter: &Limiter,
     ) -> Result<()> {
@@ -235,6 +236,7 @@ impl SSTImporter {
             src_file_name,
             dst_file.clone(),
             file_length,
+            expect_sha256,
             speed_limiter,
             file_crypter,
         );
@@ -271,12 +273,16 @@ impl SSTImporter {
         let name = meta.get_name();
         let path = self.dir.get_import_path(name)?;
         let start = Instant::now();
+        let sha256 = meta.get_sha256().to_vec();
+        let expected_sha256 = if sha256.len() > 0 { Some(sha256) } else { None };
+
         self.download_file_from_external_storage(
             // don't check file length after download file for now.
             meta.get_length(),
             name,
             path.temp.clone(),
             backend,
+            expected_sha256,
             // don't support encrypt for now.
             None,
             speed_limiter,
@@ -416,6 +422,7 @@ impl SSTImporter {
             name,
             path.temp.clone(),
             backend,
+            None,
             file_crypter,
             speed_limiter,
         )?;
@@ -709,6 +716,7 @@ mod tests {
         SeekKey, SstReader, SstWriter, CF_DEFAULT, DATA_CFS,
     };
     use file_system::File;
+    use openssl::hash::{Hasher, MessageDigest};
     use tempfile::Builder;
     use test_sst_importer::*;
     use test_util::new_test_key_manager;
@@ -1062,11 +1070,17 @@ mod tests {
         let mut input = data;
         let mut output = Vec::new();
         let input_len = input.len() as u64;
+
+        let mut hasher = Hasher::new(MessageDigest::sha256()).unwrap();
+        hasher.update(data);
+        let hash256 = hasher.finish().unwrap().to_vec();
+
         block_on_external_io(external_storage_export::read_external_storage_into_file(
             &mut input,
             &mut output,
             &Limiter::new(f64::INFINITY),
             input_len,
+            Some(hash256),
             8192,
         ))
         .unwrap();
@@ -1084,6 +1098,7 @@ mod tests {
             &mut output,
             &Limiter::new(f64::INFINITY),
             0,
+            None,
             usize::MAX,
         ))
         .unwrap_err();

src/import/sst_service.rs

@@ -526,7 +526,7 @@ where
                             import_err
                                 .set_message("failed to complete raft command".to_string());
                             // FIXME: if there are many errors, we may lose some of them here.
-                            import_err 
+                            import_err
                                 .set_store_error(err.clone());
                             warn!("failed to apply the file to the store"; "error" => ?err, "file" => %meta.get_name());
                             resp.set_error(import_err);