Move snapshot archive generation out of the SnapshotPackagerService

ledger/Cargo.toml

@@ -43,6 +43,7 @@ solana-sdk = { path = "../sdk", version = "0.23.0" }
 solana-stake-program = { path = "../programs/stake", version = "0.23.0" }
 solana-vote-program = { path = "../programs/vote", version = "0.23.0" }
 sys-info = "0.5.8"
+symlink = "0.1.0"
 tar = "0.4.26"
 thiserror = "1.0"
 tempfile = "3.1.0"

ledger/src/snapshot_utils.rs

@@ -12,10 +12,10 @@ use solana_sdk::transaction::Result as TransactionResult;
 use solana_sdk::{clock::Slot, transaction};
 use std::{
     cmp::Ordering,
-    fs,
-    fs::File,
+    fs::{self, File},
     io::{BufReader, BufWriter, Error as IOError, ErrorKind, Read, Seek, SeekFrom, Write},
     path::{Path, PathBuf},
+    process::ExitStatus,
 };
 use tar::Archive;
 use tempfile::TempDir;
@@ -42,6 +42,12 @@ pub enum SnapshotError {
 
     #[error("file system error")]
     FsExtra(#[from] fs_extra::error::Error),
+
+    #[error("archive generation failure {0}")]
+    ArchiveGenerationFailure(ExitStatus),
+
+    #[error("storage path symlink is invalid")]
+    StoragePathSymlinkInvalid,
 }
 pub type Result<T> = std::result::Result<T, SnapshotError>;
 
@@ -107,6 +113,110 @@ pub fn package_snapshot<P: AsRef<Path>, Q: AsRef<Path>>(
     Ok(package)
 }
 
+pub fn archive_snapshot_package(snapshot_package: &SnapshotPackage) -> Result<()> {
+    info!(
+        "Generating snapshot archive for slot {}",
+        snapshot_package.root
+    );
+
+    serialize_status_cache(
+        snapshot_package.root,
+        &snapshot_package.slot_deltas,
+        &snapshot_package.snapshot_links,
+    )?;
+
+    let mut timer = Measure::start("snapshot_package-package_snapshots");
+    let tar_dir = snapshot_package
+        .tar_output_file
+        .parent()
+        .expect("Tar output path is invalid");
+
+    fs::create_dir_all(tar_dir)?;
+
+    // Create the staging directories
+    let staging_dir = TempDir::new()?;
+    let staging_accounts_dir = staging_dir.path().join(TAR_ACCOUNTS_DIR);
+    let staging_snapshots_dir = staging_dir.path().join(TAR_SNAPSHOTS_DIR);
+    let staging_version_file = staging_dir.path().join(TAR_VERSION_FILE);
+    fs::create_dir_all(&staging_accounts_dir)?;
+
+    // Add the snapshots to the staging directory
+    symlink::symlink_dir(
+        snapshot_package.snapshot_links.path(),
+        &staging_snapshots_dir,
+    )?;
+
+    // Add the AppendVecs into the compressible list
+    for storage in &snapshot_package.storage_entries {
+        storage.flush()?;
+        let storage_path = storage.get_path();
+        let output_path = staging_accounts_dir.join(
+            storage_path
+                .file_name()
+                .expect("Invalid AppendVec file path"),
+        );
+
+        // `storage_path` - The file path where the AppendVec itself is located
+        // `output_path` - The directory where the AppendVec will be placed in the staging directory.
+        let storage_path =
+            fs::canonicalize(storage_path).expect("Could not get absolute path for accounts");
+        symlink::symlink_dir(storage_path, &output_path)?;
+        if !output_path.is_file() {
+            return Err(SnapshotError::StoragePathSymlinkInvalid);
+        }
+    }
+
+    // Write version file
+    {
+        let snapshot_version = format!("{}\n", env!("CARGO_PKG_VERSION"));
+        let mut f = std::fs::File::create(staging_version_file)?;
+        //f.write_all(&snapshot_version.to_string().into_bytes())?;
+        f.write_all(&snapshot_version.into_bytes())?;
+    }
+
+    // Tar the staging directory into the archive at `archive_path`
+    let archive_path = tar_dir.join("new_state.tar.bz2");
+    let args = vec![
+        "jcfhS",
+        archive_path.to_str().unwrap(),
+        "-C",
+        staging_dir.path().to_str().unwrap(),
+        TAR_ACCOUNTS_DIR,
+        TAR_SNAPSHOTS_DIR,
+        TAR_VERSION_FILE,
+    ];
+
+    let output = std::process::Command::new("tar").args(&args).output()?;
+    if !output.status.success() {
+        warn!("tar command failed with exit code: {}", output.status);
+        use std::str::from_utf8;
+        info!("tar stdout: {}", from_utf8(&output.stdout).unwrap_or("?"));
+        info!("tar stderr: {}", from_utf8(&output.stderr).unwrap_or("?"));
+
+        return Err(SnapshotError::ArchiveGenerationFailure(output.status));
+    }
+
+    // Once everything is successful, overwrite the previous tarball so that other validators
+    // can fetch this newly packaged snapshot
+    let metadata = fs::metadata(&archive_path)?;
+    fs::rename(&archive_path, &snapshot_package.tar_output_file)?;
+
+    timer.stop();
+    info!(
+        "Successfully created tarball. slot: {}, elapsed ms: {}, size={}",
+        snapshot_package.root,
+        timer.as_ms(),
+        metadata.len()
+    );
+    datapoint_info!(
+        "snapshot-package",
+        ("slot", snapshot_package.root, i64),
+        ("duration_ms", timer.as_ms(), i64),
+        ("size", metadata.len(), i64)
+    );
+    Ok(())
+}
+
 pub fn get_snapshot_paths<P: AsRef<Path>>(snapshot_path: P) -> Vec<SlotSnapshotPaths>
 where
     P: std::fmt::Debug,
@@ -469,8 +579,11 @@ fn get_io_error(error: &str) -> SnapshotError {
     SnapshotError::IO(IOError::new(ErrorKind::Other, error))
 }
 
-pub fn verify_snapshot_tar<P, Q, R>(snapshot_tar: P, snapshots_to_verify: Q, storages_to_verify: R)
-where
+pub fn verify_snapshot_archive<P, Q, R>(
+    snapshot_tar: P,
+    snapshots_to_verify: Q,
+    storages_to_verify: R,
+) where
     P: AsRef<Path>,
     Q: AsRef<Path>,
     R: AsRef<Path>,