From 21eec80d60cb951031ec5fbdaf4537acaf189f0f Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Sun, 31 May 2026 17:42:40 +0800 Subject: [PATCH] feat: add safe vtl archive pipeline Add a coldstore-vtl crate with an in-memory VTL simulator and safe command-spec/discovery helpers for mhVTL preparation.\n\nBridge the simulator into the tape service behind a TapeBackend abstraction, then wire scheduler archive batching through CacheArchiveClient and TapeArchiveClient adapters. The archive path is covered by in-process metadata/cache/tape gRPC tests and remains free of live mhVTL or host /dev access.\n\nUpdate safe verification targets and docs to describe the current Phase-1 archive pipeline. --- Cargo.toml | 1 + Makefile | 12 +- README_RUST.md | 3 +- crates/scheduler/Cargo.toml | 2 + crates/scheduler/src/service.rs | 899 ++++++++++++++++++++++++- crates/tape/Cargo.toml | 1 + crates/tape/src/service.rs | 616 +++++++++++++++-- crates/tape/tests/simulator_service.rs | 179 +++++ crates/vtl/Cargo.toml | 11 + crates/vtl/README.md | 51 ++ crates/vtl/src/command.rs | 146 ++++ crates/vtl/src/discover.rs | 144 ++++ crates/vtl/src/error.rs | 50 ++ crates/vtl/src/interface.rs | 20 + crates/vtl/src/lib.rs | 22 + crates/vtl/src/mhvtl.rs | 208 ++++++ crates/vtl/src/model.rs | 223 ++++++ crates/vtl/src/simulator.rs | 291 ++++++++ crates/vtl/tests/behavior.rs | 152 +++++ docs/modules/05-scheduler-layer.md | 11 + docs/modules/06-tape-layer.md | 13 + scripts/setup-mhvtl-env.sh | 179 +++++ 22 files changed, 3170 insertions(+), 64 deletions(-) create mode 100644 crates/tape/tests/simulator_service.rs create mode 100644 crates/vtl/Cargo.toml create mode 100644 crates/vtl/README.md create mode 100644 crates/vtl/src/command.rs create mode 100644 crates/vtl/src/discover.rs create mode 100644 crates/vtl/src/error.rs create mode 100644 crates/vtl/src/interface.rs create mode 100644 crates/vtl/src/lib.rs create mode 100644 crates/vtl/src/mhvtl.rs create mode 100644 crates/vtl/src/model.rs create mode 100644 crates/vtl/src/simulator.rs create mode 100644 crates/vtl/tests/behavior.rs create mode 100755 scripts/setup-mhvtl-env.sh diff --git a/Cargo.toml b/Cargo.toml index 6391457..44f3949 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "crates/scheduler", "crates/cache", "crates/tape", + "crates/vtl", ] [workspace.package] diff --git a/Makefile b/Makefile index ddce59d..55c1b50 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: build build-debug fmt fmt-check clippy test unit test-unit check check-all check-safe clean help install-hooks install-tools setup lint run-metadata run-gateway run-scheduler run-cache run-tape +.PHONY: build build-debug fmt fmt-check clippy test unit test-unit vtl-unit vtl-check check check-all check-safe clean help install-hooks install-tools setup lint run-metadata run-gateway run-scheduler run-cache run-tape .DEFAULT_GOAL := help @@ -21,11 +21,15 @@ test: @cargo test --workspace --all-features unit test-unit: @cargo test --workspace --lib --bins +vtl-unit: + @cargo test -p coldstore-vtl --lib --tests +vtl-check: + @cargo check -p coldstore-vtl check: @cargo check --all-targets --all-features check-all: fmt-check clippy test -check-safe: fmt-check clippy unit build-debug - @echo "Safe verification passed (fmt/clippy/unit/build only)." +check-safe: fmt-check clippy unit vtl-unit build-debug + @echo "Safe verification passed (fmt/clippy/unit/vtl-unit/build only)." clean: @cargo clean @@ -59,6 +63,6 @@ help: @echo "ColdStore Workspace Makefile" @echo "" @echo "Build: build build-debug clean" - @echo "Quality: fmt fmt-check clippy test unit check check-all check-safe lint" + @echo "Quality: fmt fmt-check clippy test unit vtl-unit vtl-check check check-all check-safe lint" @echo "Run: run-metadata run-gateway run-scheduler run-cache run-tape" @echo "Setup: setup install-tools install-hooks" diff --git a/README_RUST.md b/README_RUST.md index 7d17c91..bb9c418 100644 --- a/README_RUST.md +++ b/README_RUST.md @@ -15,7 +15,7 @@ ColdStore 是**纯冷归档系统**（类似 AWS Glacier Deep Archive）： | 分期 | 状态 | 已落地能力 | |------|------|------------| -| Phase 1 | 已落地 / 可单测 | Gateway/Scheduler/Metadata/Cache 本地闭环；bucket/object CRUD；Put/Head/Get/Delete/Restore/List；HDD Cache staging/restored；Phase-1 archive 标记 Cold 并清理 staging | +| Phase 1 | 已落地 / 可单测 | Gateway/Scheduler/Metadata/Cache 本地闭环；bucket/object CRUD；Put/Head/Get/Delete/Restore/List；HDD Cache staging/restored；Phase-1 archive 标记 Cold 并清理 staging；TapeService + coldstore-vtl 安全模拟驱动/带库/load/unload/filemark 读写闭环；Scheduler archive batch 可注入 cache+tape writer 并写入 ArchiveBundle/filemark 元数据；CacheArchiveClient 已接入真实 CacheService staging list/get/delete 单测；TapeArchiveClient 已接入真实 TapeService WriteBundle/ReadBundle gRPC 单测 | | Phase 2A | 已启动 / 可单测 | Metadata opt-in 二进制 snapshot：`MetadataServiceImpl::new_with_snapshot(config, path)` 支持写入后保存、重启后恢复 bucket/object/task/worker/tape 等状态 | | Phase 2B | 下一步 | 将 Metadata 状态机接入 OpenRaft + RocksDB/openraft-rocksstore，补齐安全的 Raft 状态机单测和多节点一致性测试 | @@ -32,6 +32,7 @@ ColdStore 是**纯冷归档系统**（类似 AWS Glacier Deep Archive）： | coldstore-scheduler | bin | 调度 Worker（业务中枢） | | coldstore-cache | bin | 缓存 Worker（独立进程，HDD/SPDK） | | coldstore-tape | bin | 磁带 Worker（独立物理节点） | +| coldstore-vtl | lib | mhVTL/虚拟磁带库 harness：lsscsi/mtx/mt/sg 命令封装 + 安全内存模拟器 | ## 组件间通信 diff --git a/crates/scheduler/Cargo.toml b/crates/scheduler/Cargo.toml index b68ef4c..27a4a98 100644 --- a/crates/scheduler/Cargo.toml +++ b/crates/scheduler/Cargo.toml @@ -29,4 +29,6 @@ tokio-stream = { workspace = true } prost-types = { workspace = true } [dev-dependencies] +coldstore-cache = { path = "../cache" } coldstore-metadata = { path = "../metadata" } +coldstore-tape = { path = "../tape" } diff --git a/crates/scheduler/src/service.rs b/crates/scheduler/src/service.rs index f53db75..04a9a3f 100644 --- a/crates/scheduler/src/service.rs +++ b/crates/scheduler/src/service.rs @@ -1,7 +1,18 @@ use crate::SchedulerState; +use coldstore_proto::cache::cache_service_client::CacheServiceClient; +use coldstore_proto::cache::get_staging_response::Payload as GetStagingPayload; +use coldstore_proto::cache::{ + DeleteStagingRequest, GetStagingRequest, ListStagingKeysRequest, StagingKeyEntry, + StagingObjectMeta, +}; use coldstore_proto::common; use coldstore_proto::scheduler::scheduler_service_server::SchedulerService; use coldstore_proto::scheduler::*; +use coldstore_proto::tape::tape_service_client::TapeServiceClient as TapeGrpcClient; +use coldstore_proto::tape::write_bundle_request::Payload as TapeWriteRequestPayload; +use coldstore_proto::tape::{ + WriteBundleMeta as TapeWriteBundleMeta, WriteBundleRequest as TapeWriteBundleRequest, +}; use prost_types::Timestamp; use sha2::{Digest, Sha256}; #[cfg(test)] @@ -53,20 +64,347 @@ pub trait Phase1SchedulerBackend: Send + Sync + 'static { ) -> std::result::Result, Status>; } -struct MetadataBackedSchedulerBackend { +#[derive(Debug, Clone)] +pub struct Phase1StagedObject { + pub meta: StagingObjectMeta, + pub data: Vec, +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct TapeArchiveWrite { + pub tape_id: String, + pub tape_set: Vec, + pub filemark_start: u32, + pub filemark_end: u32, + pub bytes_written: u64, +} + +#[derive(Debug, Clone, Default, Eq, PartialEq)] +pub struct ArchiveBatchResult { + pub archived_objects: u32, + pub bytes_written: u64, + pub bundle_ids: Vec, +} + +#[tonic::async_trait] +pub trait Phase1ArchiveCache: Send + Sync { + async fn list_staging_keys( + &self, + limit: u32, + ) -> std::result::Result, Status>; + + async fn get_staging( + &self, + bucket: &str, + key: &str, + version_id: Option<&str>, + ) -> std::result::Result; + + async fn delete_staging( + &self, + bucket: &str, + key: &str, + version_id: Option<&str>, + ) -> std::result::Result<(), Status>; +} + +#[derive(Clone)] +pub struct CacheArchiveClient { + client: CacheServiceClient, +} + +impl CacheArchiveClient { + pub fn new(client: CacheServiceClient) -> Self { + Self { client } + } +} + +#[tonic::async_trait] +impl Phase1ArchiveCache for CacheArchiveClient { + async fn list_staging_keys( + &self, + limit: u32, + ) -> std::result::Result, Status> { + let mut client = self.client.clone(); + Ok(client + .list_staging_keys(Request::new(ListStagingKeysRequest { limit, after: None })) + .await? + .into_inner() + .entries) + } + + async fn get_staging( + &self, + bucket: &str, + key: &str, + version_id: Option<&str>, + ) -> std::result::Result { + let mut client = self.client.clone(); + let mut stream = client + .get_staging(Request::new(GetStagingRequest { + bucket: bucket.into(), + key: key.into(), + version_id: version_id.map(str::to_owned), + })) + .await? + .into_inner(); + + let mut meta = None; + let mut data = Vec::new(); + while let Some(message) = stream.message().await? { + match message.payload { + Some(GetStagingPayload::Meta(next_meta)) => { + if meta.replace(next_meta).is_some() { + return Err(Status::invalid_argument( + "cache get_staging returned duplicate metadata", + )); + } + } + Some(GetStagingPayload::Data(chunk)) => data.extend_from_slice(&chunk), + None => return Err(Status::internal("cache get_staging returned empty chunk")), + } + } + + let meta = meta + .ok_or_else(|| Status::internal("cache get_staging stream ended without metadata"))?; + Ok(Phase1StagedObject { meta, data }) + } + + async fn delete_staging( + &self, + bucket: &str, + key: &str, + version_id: Option<&str>, + ) -> std::result::Result<(), Status> { + let mut client = self.client.clone(); + client + .delete_staging(Request::new(DeleteStagingRequest { + bucket: bucket.into(), + key: key.into(), + version_id: version_id.map(str::to_owned), + })) + .await?; + Ok(()) + } +} + +#[tonic::async_trait] +pub trait TapeArchiveWriter: Send + Sync { + async fn write_bundle( + &self, + bundle_id: &str, + object_count: u32, + data: Vec, + ) -> std::result::Result; +} + +#[derive(Clone)] +pub struct TapeArchiveClient { + client: TapeGrpcClient, + drive_id: String, + tape_id: String, + tape_set: Vec, + block_size: u32, +} + +impl TapeArchiveClient { + pub fn new( + client: TapeGrpcClient, + drive_id: impl Into, + tape_id: impl Into, + tape_set: Vec, + block_size: u32, + ) -> Self { + Self { + client, + drive_id: drive_id.into(), + tape_id: tape_id.into(), + tape_set, + block_size, + } + } +} + +#[tonic::async_trait] +impl TapeArchiveWriter for TapeArchiveClient { + async fn write_bundle( + &self, + bundle_id: &str, + object_count: u32, + data: Vec, + ) -> std::result::Result { + let mut client = self.client.clone(); + let response = client + .write_bundle(Request::new(tokio_stream::iter(vec![ + TapeWriteBundleRequest { + payload: Some(TapeWriteRequestPayload::Meta(TapeWriteBundleMeta { + drive_id: self.drive_id.clone(), + bundle_id: bundle_id.into(), + total_size: data.len() as u64, + object_count, + block_size: self.block_size, + })), + }, + TapeWriteBundleRequest { + payload: Some(TapeWriteRequestPayload::Data(data)), + }, + ]))) + .await? + .into_inner(); + + if !response.success { + return Err(Status::internal(format!( + "tape write_bundle failed for {bundle_id}: {}", + response + .error + .unwrap_or_else(|| "unknown tape error".into()) + ))); + } + if response.drive_id != self.drive_id { + return Err(Status::internal(format!( + "tape write_bundle responded for drive {}, expected {}", + response.drive_id, self.drive_id + ))); + } + if response.bundle_id != bundle_id { + return Err(Status::internal(format!( + "tape write_bundle responded for bundle {}, expected {bundle_id}", + response.bundle_id + ))); + } + + Ok(TapeArchiveWrite { + tape_id: self.tape_id.clone(), + tape_set: self.tape_set.clone(), + filemark_start: response.filemark_start, + filemark_end: response.filemark_end, + bytes_written: response.bytes_written, + }) + } +} + +pub struct MetadataBackedSchedulerBackend { metadata: coldstore_proto::metadata::metadata_service_client::MetadataServiceClient< tonic::transport::Channel, >, } impl MetadataBackedSchedulerBackend { - fn new( + pub fn new( metadata: coldstore_proto::metadata::metadata_service_client::MetadataServiceClient< tonic::transport::Channel, >, ) -> Self { Self { metadata } } + + pub async fn archive_staging_batch( + &self, + cache: &C, + tape: &T, + limit: u32, + ) -> std::result::Result + where + C: Phase1ArchiveCache + ?Sized, + T: TapeArchiveWriter + ?Sized, + { + let entries = cache.list_staging_keys(limit).await?; + let mut result = ArchiveBatchResult::default(); + + for entry in entries { + let object = self.head_object(&entry.bucket, &entry.key).await?; + if object.storage_class != common::StorageClass::ColdPending as i32 { + continue; + } + + let staged = cache + .get_staging(&entry.bucket, &entry.key, entry.version_id.as_deref()) + .await?; + if staged.meta.size != staged.data.len() as u64 { + return Err(Status::invalid_argument(format!( + "staging size mismatch for {}/{}: meta={}, bytes={}", + entry.bucket, + entry.key, + staged.meta.size, + staged.data.len() + ))); + } + + let bundle_id = + phase1_bundle_id(&entry.bucket, &entry.key, entry.version_id.as_deref()); + let checksum = staged + .meta + .checksum + .clone() + .unwrap_or_else(|| sha256_hex(&staged.data)); + let write = tape + .write_bundle(&bundle_id, 1, staged.data.clone()) + .await?; + if write.bytes_written != staged.data.len() as u64 { + return Err(Status::internal(format!( + "tape writer reported {} bytes for {bundle_id}, expected {}", + write.bytes_written, + staged.data.len() + ))); + } + + let now = now_timestamp(); + let bundle = common::ArchiveBundle { + id: bundle_id.clone(), + tape_id: write.tape_id.clone(), + tape_set: write.tape_set.clone(), + entries: vec![common::BundleEntry { + bucket: entry.bucket.clone(), + key: entry.key.clone(), + version_id: entry.version_id.clone(), + size: staged.data.len() as u64, + offset_in_bundle: 0, + tape_block_offset: write.filemark_start as u64, + checksum: checksum.clone(), + }], + total_size: write.bytes_written, + filemark_start: write.filemark_start, + filemark_end: write.filemark_end, + checksum: Some(checksum), + status: common::ArchiveBundleStatus::BundleCompleted as i32, + created_at: Some(now), + completed_at: Some(now), + }; + + let mut client = self.metadata.clone(); + client.put_archive_bundle(Request::new(bundle)).await?; + client + .update_archive_location(Request::new( + coldstore_proto::metadata::UpdateArchiveLocationRequest { + bucket: entry.bucket.clone(), + key: entry.key.clone(), + archive_id: bundle_id.clone(), + tape_id: write.tape_id.clone(), + tape_set: write.tape_set.clone(), + tape_block_offset: write.filemark_start as u64, + }, + )) + .await?; + client + .update_storage_class(Request::new( + coldstore_proto::metadata::UpdateStorageClassRequest { + bucket: entry.bucket.clone(), + key: entry.key.clone(), + storage_class: common::StorageClass::Cold as i32, + }, + )) + .await?; + cache + .delete_staging(&entry.bucket, &entry.key, entry.version_id.as_deref()) + .await?; + + result.archived_objects += 1; + result.bytes_written += write.bytes_written; + result.bundle_ids.push(bundle_id); + } + + Ok(result) + } } #[tonic::async_trait] @@ -398,6 +736,13 @@ fn storage_class_label(storage_class: i32) -> &'static str { } } +fn phase1_bundle_id(bucket: &str, key: &str, version_id: Option<&str>) -> String { + match version_id.filter(|version| !version.is_empty()) { + Some(version) => format!("phase1-bundle:{bucket}/{key}#{version}"), + None => format!("phase1-bundle:{bucket}/{key}"), + } +} + #[tonic::async_trait] impl SchedulerService for SchedulerServiceImpl { async fn put_object( @@ -579,13 +924,36 @@ impl SchedulerService for SchedulerServiceImpl { #[cfg(test)] mod tests { use super::*; - use coldstore_common::config::{MetadataConfig, SchedulerConfig}; + use coldstore_cache::service::CacheServiceImpl; + use coldstore_common::config::{ + CacheBackendConfig, CacheConfig, MetadataConfig, SchedulerConfig, TapeConfig, + }; use coldstore_metadata::service::MetadataServiceImpl; + use coldstore_proto::cache::cache_service_client::CacheServiceClient; + use coldstore_proto::cache::cache_service_server::CacheServiceServer; + use coldstore_proto::cache::put_staging_request::Payload as PutStagingPayload; + use coldstore_proto::cache::{ + GetStagingRequest, ListStagingKeysRequest, PutStagingMeta, PutStagingRequest, + StagingKeyEntry, StagingObjectMeta, + }; use coldstore_proto::metadata::metadata_service_server::MetadataServiceServer; + use coldstore_proto::tape::read_bundle_request::Location as TapeReadLocation; + use coldstore_proto::tape::read_bundle_response::Payload as TapeReadPayload; + use coldstore_proto::tape::tape_service_client::TapeServiceClient; + use coldstore_proto::tape::tape_service_server::{ + TapeService as TapeWorkerService, TapeServiceServer, + }; + use coldstore_proto::tape::write_bundle_request::Payload as TapeWritePayload; + use coldstore_proto::tape::{ + LoadTapeRequest, ReadBundleRequest as TapeReadBundleRequest, WriteBundleMeta, + WriteBundleRequest, + }; + use coldstore_tape::service::{SimulatorTapeBackend, TapeServiceImpl as TapeWorkerServiceImpl}; + use std::time::{SystemTime, UNIX_EPOCH}; use tokio::sync::oneshot; use tokio::time::{sleep, Duration}; use tokio_stream::StreamExt; - use tonic::transport::Server; + use tonic::transport::{Channel, Server}; #[derive(Default)] struct InMemoryBackend { @@ -1033,6 +1401,116 @@ mod tests { (SchedulerServiceImpl::new(state.clone()), state, shutdown_tx) } + fn test_cache_config() -> CacheConfig { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("time") + .as_nanos(); + CacheConfig { + backend: CacheBackendConfig::Hdd { + path: format!("/tmp/coldstore-scheduler-cache-test-{unique}"), + max_size_gb: 1, + }, + ..CacheConfig::default() + } + } + + async fn cache_backed_service() -> (CacheServiceClient, oneshot::Sender<()>) { + let cache = CacheServiceImpl::new(&test_cache_config()) + .await + .expect("cache service init"); + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind test listener"); + let addr = listener.local_addr().expect("listener addr"); + drop(listener); + + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + tokio::spawn(async move { + Server::builder() + .add_service(CacheServiceServer::new(cache)) + .serve_with_shutdown(addr, async { + let _ = shutdown_rx.await; + }) + .await + .expect("cache server should run"); + }); + + let mut cache_client = None; + for _ in 0..20 { + match CacheServiceClient::connect(format!("http://{addr}")).await { + Ok(client) => { + cache_client = Some(client); + break; + } + Err(_) => sleep(Duration::from_millis(25)).await, + } + } + (cache_client.expect("connect cache client"), shutdown_tx) + } + + async fn tape_backed_service() -> (TapeServiceClient, oneshot::Sender<()>) { + let backend = SimulatorTapeBackend::new(2, 1); + backend.insert_tape("slot-1", "TAPE-GRPC").unwrap(); + let tape = TapeWorkerServiceImpl::new_with_backend(TapeConfig::default(), backend); + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind test listener"); + let addr = listener.local_addr().expect("listener addr"); + drop(listener); + + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + tokio::spawn(async move { + Server::builder() + .add_service(TapeServiceServer::new(tape)) + .serve_with_shutdown(addr, async { + let _ = shutdown_rx.await; + }) + .await + .expect("tape server should run"); + }); + + let mut tape_client = None; + for _ in 0..20 { + match TapeServiceClient::connect(format!("http://{addr}")).await { + Ok(client) => { + tape_client = Some(client); + break; + } + Err(_) => sleep(Duration::from_millis(25)).await, + } + } + let mut tape_client = tape_client.expect("connect tape client"); + tape_client + .load_tape(Request::new(LoadTapeRequest { + tape_id: "TAPE-GRPC".into(), + drive_id: "drive-0".into(), + slot_id: Some("slot-1".into()), + })) + .await + .expect("load tape into in-process simulator"); + (tape_client, shutdown_tx) + } + + async fn read_tape_client_filemark( + client: &mut TapeServiceClient, + filemark: u32, + length: u64, + ) -> Vec { + let mut stream = client + .read_bundle(Request::new(TapeReadBundleRequest { + drive_id: "drive-0".into(), + location: Some(TapeReadLocation::Filemark(filemark)), + length, + })) + .await + .expect("read bundle through tape client") + .into_inner(); + while let Some(message) = stream.next().await { + match message.unwrap().payload.unwrap() { + TapeReadPayload::Meta(_) => {} + TapeReadPayload::Data(data) => return data, + } + } + Vec::new() + } + #[tokio::test] async fn default_service_uses_metadata_for_bucket_ops() { let (svc, _state, shutdown_tx) = metadata_backed_service().await; @@ -1218,6 +1696,419 @@ mod tests { shutdown_tx.send(()).ok(); } + struct TestArchiveCache { + staged: RwLock>, + deleted: RwLock>, + } + + impl TestArchiveCache { + fn with_object(bucket: &str, key: &str, data: Vec) -> Self { + let meta = StagingObjectMeta { + bucket: bucket.into(), + key: key.into(), + version_id: None, + size: data.len() as u64, + checksum: Some(sha256_hex(&data)), + content_type: Some("text/plain".into()), + etag: Some("etag-staged".into()), + staged_at: Some(Timestamp { + seconds: 30, + nanos: 0, + }), + }; + let mut staged = HashMap::new(); + staged.insert(format!("{bucket}/{key}"), Phase1StagedObject { meta, data }); + Self { + staged: RwLock::new(staged), + deleted: RwLock::new(Vec::new()), + } + } + + fn deleted_keys(&self) -> Vec { + self.deleted.read().unwrap().clone() + } + } + + #[tonic::async_trait] + impl Phase1ArchiveCache for TestArchiveCache { + async fn list_staging_keys( + &self, + limit: u32, + ) -> std::result::Result, Status> { + let mut entries: Vec<_> = self + .staged + .read() + .unwrap() + .values() + .map(|object| StagingKeyEntry { + bucket: object.meta.bucket.clone(), + key: object.meta.key.clone(), + version_id: object.meta.version_id.clone(), + size: object.meta.size, + staged_at: object.meta.staged_at, + }) + .collect(); + entries.sort_by(|a, b| (&a.bucket, &a.key).cmp(&(&b.bucket, &b.key))); + entries.truncate(limit as usize); + Ok(entries) + } + + async fn get_staging( + &self, + bucket: &str, + key: &str, + _version_id: Option<&str>, + ) -> std::result::Result { + self.staged + .read() + .unwrap() + .get(&format!("{bucket}/{key}")) + .cloned() + .ok_or_else(|| Status::not_found("staging object missing")) + } + + async fn delete_staging( + &self, + bucket: &str, + key: &str, + _version_id: Option<&str>, + ) -> std::result::Result<(), Status> { + self.staged + .write() + .unwrap() + .remove(&format!("{bucket}/{key}")); + self.deleted + .write() + .unwrap() + .push(format!("{bucket}/{key}")); + Ok(()) + } + } + + struct DirectTapeWriter { + service: TapeWorkerServiceImpl, + } + + impl DirectTapeWriter { + async fn loaded() -> Self { + let backend = SimulatorTapeBackend::new(2, 1); + backend.insert_tape("slot-1", "TAPE-PHASE1").unwrap(); + let service = TapeWorkerServiceImpl::new_with_backend(TapeConfig::default(), backend); + service + .load_tape(Request::new(LoadTapeRequest { + tape_id: "TAPE-PHASE1".into(), + drive_id: "drive-0".into(), + slot_id: Some("slot-1".into()), + })) + .await + .unwrap(); + Self { service } + } + + async fn read_filemark(&self, filemark: u32, length: u64) -> Vec { + let mut stream = self + .service + .read_bundle(Request::new(TapeReadBundleRequest { + drive_id: "drive-0".into(), + location: Some(TapeReadLocation::Filemark(filemark)), + length, + })) + .await + .unwrap() + .into_inner(); + while let Some(message) = stream.next().await { + match message.unwrap().payload.unwrap() { + TapeReadPayload::Meta(_) => {} + TapeReadPayload::Data(data) => return data, + } + } + Vec::new() + } + } + + #[tonic::async_trait] + impl TapeArchiveWriter for DirectTapeWriter { + async fn write_bundle( + &self, + bundle_id: &str, + object_count: u32, + data: Vec, + ) -> std::result::Result { + let response = self + .service + .write_bundle_from_messages(tokio_stream::iter(vec![ + Ok(WriteBundleRequest { + payload: Some(TapeWritePayload::Meta(WriteBundleMeta { + drive_id: "drive-0".into(), + bundle_id: bundle_id.into(), + total_size: data.len() as u64, + object_count, + block_size: 262_144, + })), + }), + Ok(WriteBundleRequest { + payload: Some(TapeWritePayload::Data(data)), + }), + ])) + .await?; + Ok(TapeArchiveWrite { + tape_id: "TAPE-PHASE1".into(), + tape_set: vec!["TAPE-PHASE1".into()], + filemark_start: response.filemark_start, + filemark_end: response.filemark_end, + bytes_written: response.bytes_written, + }) + } + } + + #[tokio::test] + async fn archive_staging_batch_writes_tape_and_updates_metadata() { + let (_svc, state, shutdown_tx) = metadata_backed_service().await; + let backend = MetadataBackedSchedulerBackend::new(state.metadata.clone()); + + backend + .create_bucket("docs") + .await + .expect("create bucket through metadata backend"); + backend + .put_object( + "docs", + "guide.txt", + b"abcdef".to_vec(), + Some("text/plain".into()), + ) + .await + .expect("put object metadata"); + + let cache = TestArchiveCache::with_object("docs", "guide.txt", b"abcdef".to_vec()); + let tape = DirectTapeWriter::loaded().await; + + let archived = backend + .archive_staging_batch(&cache, &tape, 10) + .await + .expect("archive staging batch"); + assert_eq!(archived.archived_objects, 1); + assert_eq!(archived.bytes_written, 6); + assert_eq!(archived.bundle_ids, vec!["phase1-bundle:docs/guide.txt"]); + + let object = backend + .head_object("docs", "guide.txt") + .await + .expect("head archived object"); + assert_eq!(object.storage_class, common::StorageClass::Cold as i32); + assert_eq!( + object.archive_id.as_deref(), + Some("phase1-bundle:docs/guide.txt") + ); + assert_eq!(object.tape_id.as_deref(), Some("TAPE-PHASE1")); + assert_eq!(object.tape_set, vec!["TAPE-PHASE1"]); + assert_eq!(object.tape_block_offset, Some(0)); + + let mut metadata = state.metadata.clone(); + let bundle = metadata + .get_archive_bundle(Request::new( + coldstore_proto::metadata::GetArchiveBundleRequest { + id: "phase1-bundle:docs/guide.txt".into(), + }, + )) + .await + .expect("archive bundle stored") + .into_inner(); + assert_eq!(bundle.tape_id, "TAPE-PHASE1"); + assert_eq!(bundle.filemark_start, 0); + assert_eq!(bundle.filemark_end, 1); + assert_eq!(bundle.total_size, 6); + assert_eq!(bundle.entries.len(), 1); + assert_eq!(bundle.entries[0].bucket, "docs"); + assert_eq!(bundle.entries[0].key, "guide.txt"); + assert_eq!(bundle.entries[0].tape_block_offset, 0); + + assert_eq!(cache.deleted_keys(), vec!["docs/guide.txt"]); + assert_eq!(tape.read_filemark(0, 6).await, b"abcdef"); + + shutdown_tx.send(()).ok(); + } + + #[tokio::test] + async fn archive_staging_batch_consumes_real_cache_service_staging() { + let (_svc, state, metadata_shutdown_tx) = metadata_backed_service().await; + let backend = MetadataBackedSchedulerBackend::new(state.metadata.clone()); + + backend + .create_bucket("docs") + .await + .expect("create bucket through metadata backend"); + backend + .put_object( + "docs", + "from-cache.txt", + b"cache-body".to_vec(), + Some("text/plain".into()), + ) + .await + .expect("put object metadata"); + + let (mut raw_cache_client, cache_shutdown_tx) = cache_backed_service().await; + let body = b"cache-body".to_vec(); + raw_cache_client + .put_staging(Request::new(tokio_stream::iter(vec![ + PutStagingRequest { + payload: Some(PutStagingPayload::Meta(PutStagingMeta { + bucket: "docs".into(), + key: "from-cache.txt".into(), + version_id: None, + size: body.len() as u64, + checksum: Some(sha256_hex(&body)), + content_type: Some("text/plain".into()), + etag: Some("etag-from-cache".into()), + })), + }, + PutStagingRequest { + payload: Some(PutStagingPayload::Data(body.clone())), + }, + ]))) + .await + .expect("put staging through real cache service"); + + let cache = CacheArchiveClient::new(raw_cache_client.clone()); + let tape = DirectTapeWriter::loaded().await; + + let archived = backend + .archive_staging_batch(&cache, &tape, 10) + .await + .expect("archive real cache staging batch"); + assert_eq!(archived.archived_objects, 1); + assert_eq!(archived.bytes_written, body.len() as u64); + assert_eq!( + archived.bundle_ids, + vec!["phase1-bundle:docs/from-cache.txt"] + ); + + let object = backend + .head_object("docs", "from-cache.txt") + .await + .expect("head archived object"); + assert_eq!(object.storage_class, common::StorageClass::Cold as i32); + assert_eq!( + object.archive_id.as_deref(), + Some("phase1-bundle:docs/from-cache.txt") + ); + + let listed = raw_cache_client + .list_staging_keys(Request::new(ListStagingKeysRequest { + limit: 10, + after: None, + })) + .await + .expect("list staging after archive") + .into_inner(); + assert!(listed.entries.is_empty()); + + let missing = raw_cache_client + .get_staging(Request::new(GetStagingRequest { + bucket: "docs".into(), + key: "from-cache.txt".into(), + version_id: None, + })) + .await + .expect_err("staging should be deleted after archive"); + assert_eq!(missing.code(), tonic::Code::NotFound); + assert_eq!(tape.read_filemark(0, body.len() as u64).await, body); + + metadata_shutdown_tx.send(()).ok(); + cache_shutdown_tx.send(()).ok(); + } + + #[tokio::test] + async fn archive_staging_batch_uses_cache_and_tape_grpc_clients() { + let (_svc, state, metadata_shutdown_tx) = metadata_backed_service().await; + let backend = MetadataBackedSchedulerBackend::new(state.metadata.clone()); + + backend + .create_bucket("docs") + .await + .expect("create bucket through metadata backend"); + let body = b"grpc-tape-body".to_vec(); + backend + .put_object( + "docs", + "grpc-tape.txt", + body.clone(), + Some("text/plain".into()), + ) + .await + .expect("put object metadata"); + + let (mut raw_cache_client, cache_shutdown_tx) = cache_backed_service().await; + raw_cache_client + .put_staging(Request::new(tokio_stream::iter(vec![ + PutStagingRequest { + payload: Some(PutStagingPayload::Meta(PutStagingMeta { + bucket: "docs".into(), + key: "grpc-tape.txt".into(), + version_id: None, + size: body.len() as u64, + checksum: Some(sha256_hex(&body)), + content_type: Some("text/plain".into()), + etag: Some("etag-grpc-tape".into()), + })), + }, + PutStagingRequest { + payload: Some(PutStagingPayload::Data(body.clone())), + }, + ]))) + .await + .expect("put staging through real cache service"); + + let (mut raw_tape_client, tape_shutdown_tx) = tape_backed_service().await; + let cache = CacheArchiveClient::new(raw_cache_client.clone()); + let tape = TapeArchiveClient::new( + raw_tape_client.clone(), + "drive-0", + "TAPE-GRPC", + vec!["TAPE-GRPC".into()], + 262_144, + ); + + let archived = backend + .archive_staging_batch(&cache, &tape, 10) + .await + .expect("archive through cache and tape grpc clients"); + assert_eq!(archived.archived_objects, 1); + assert_eq!(archived.bytes_written, body.len() as u64); + assert_eq!( + archived.bundle_ids, + vec!["phase1-bundle:docs/grpc-tape.txt"] + ); + + let object = backend + .head_object("docs", "grpc-tape.txt") + .await + .expect("head archived object"); + assert_eq!(object.storage_class, common::StorageClass::Cold as i32); + assert_eq!(object.tape_id.as_deref(), Some("TAPE-GRPC")); + assert_eq!(object.tape_set, vec!["TAPE-GRPC"]); + assert_eq!(object.tape_block_offset, Some(0)); + + let listed = raw_cache_client + .list_staging_keys(Request::new(ListStagingKeysRequest { + limit: 10, + after: None, + })) + .await + .expect("list staging after archive") + .into_inner(); + assert!(listed.entries.is_empty()); + assert_eq!( + read_tape_client_filemark(&mut raw_tape_client, 0, body.len() as u64).await, + body + ); + + metadata_shutdown_tx.send(()).ok(); + cache_shutdown_tx.send(()).ok(); + tape_shutdown_tx.send(()).ok(); + } + #[test] fn phase1_unimplemented_message_is_stable() { let status = phase1_unimplemented("scheduler.list_buckets"); diff --git a/crates/tape/Cargo.toml b/crates/tape/Cargo.toml index f21fcce..166867b 100644 --- a/crates/tape/Cargo.toml +++ b/crates/tape/Cargo.toml @@ -12,6 +12,7 @@ path = "src/main.rs" [dependencies] coldstore-proto = { workspace = true } coldstore-common = { workspace = true } +coldstore-vtl = { path = "../vtl" } tokio = { workspace = true } tonic = { workspace = true } prost = { workspace = true } diff --git a/crates/tape/src/service.rs b/crates/tape/src/service.rs index 86f6a90..7b9ad1f 100644 --- a/crates/tape/src/service.rs +++ b/crates/tape/src/service.rs @@ -1,113 +1,619 @@ +#![allow(clippy::result_large_err)] + +use std::sync::{Arc, Mutex, MutexGuard}; + use coldstore_common::config::TapeConfig; -use coldstore_proto::common; +use coldstore_proto::common::{self, DriveStatus, TapeStatus}; +use coldstore_proto::tape::read_bundle_request::Location; +use coldstore_proto::tape::read_bundle_response::Payload as ReadPayload; use coldstore_proto::tape::tape_service_server::TapeService; +use coldstore_proto::tape::write_bundle_request::Payload as WritePayload; use coldstore_proto::tape::*; +use coldstore_vtl::model::{ElementAddress, TapeBarcode}; +use coldstore_vtl::simulator::VirtualTapeLibrary; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; +use tokio_stream::{Stream, StreamExt}; use tonic::{Request, Response, Status, Streaming}; +const SIMULATED_TAPE_CAPACITY_BYTES: u64 = 12 * 1024 * 1024 * 1024 * 1024; + +type ServiceResult = std::result::Result; + +pub trait TapeBackend: Send + Sync { + fn list_drives(&self) -> ServiceResult>; + fn get_drive_status(&self, drive_id: &str) -> ServiceResult; + fn acquire_drive( + &self, + preferred_drive_id: Option<&str>, + required_tape_id: Option<&str>, + ) -> ServiceResult; + fn release_drive(&self, drive_id: &str) -> ServiceResult<()>; + fn load_tape(&self, tape_id: &str, drive_id: &str, slot_id: Option<&str>) -> ServiceResult<()>; + fn unload_tape(&self, drive_id: &str, target_slot_id: Option<&str>) -> ServiceResult<()>; + fn rewind(&self, drive_id: &str) -> ServiceResult<()>; + fn seek_to_filemark(&self, drive_id: &str, filemark: u32) -> ServiceResult<()>; + fn get_tape_media_status(&self, drive_id: &str) -> ServiceResult; + fn inventory(&self) -> ServiceResult; + fn write_bundle(&self, drive_id: &str, data: &[u8]) -> ServiceResult<(u32, u32)>; + fn read_bundle(&self, drive_id: &str, filemark: u32, length: u64) -> ServiceResult>; +} + pub struct TapeServiceImpl { _config: TapeConfig, + backend: Arc, } impl TapeServiceImpl { pub fn new(config: &TapeConfig) -> anyhow::Result { - Ok(Self { - _config: config.clone(), + let backend = SimulatorTapeBackend::from_config(config); + Ok(Self::new_with_backend(config.clone(), backend)) + } + + pub fn new_with_backend(config: TapeConfig, backend: B) -> Self + where + B: TapeBackend + 'static, + { + Self { + _config: config, + backend: Arc::new(backend), + } + } + + pub async fn write_bundle_from_messages( + &self, + mut messages: S, + ) -> ServiceResult + where + S: Stream> + Unpin, + { + let first = match messages.next().await { + Some(Ok(request)) => request, + Some(Err(status)) => return Err(status), + None => return Err(Status::invalid_argument("write_bundle stream is empty")), + }; + let meta = match first.payload { + Some(WritePayload::Meta(meta)) => meta, + Some(WritePayload::Data(_)) => { + return Err(Status::invalid_argument( + "first write_bundle message must carry metadata", + )) + } + None => { + return Err(Status::invalid_argument( + "write_bundle message has no payload", + )) + } + }; + + let mut data = Vec::new(); + while let Some(message) = messages.next().await { + match message?.payload { + Some(WritePayload::Data(chunk)) => data.extend_from_slice(&chunk), + Some(WritePayload::Meta(_)) => { + return Err(Status::invalid_argument( + "write_bundle metadata must appear only once as the first message", + )) + } + None => { + return Err(Status::invalid_argument( + "write_bundle message has no payload", + )) + } + } + } + + if meta.total_size != data.len() as u64 { + return Err(Status::invalid_argument(format!( + "write_bundle total_size={} does not match received bytes={}", + meta.total_size, + data.len() + ))); + } + + let (filemark_start, filemark_end) = self.backend.write_bundle(&meta.drive_id, &data)?; + Ok(WriteBundleResponse { + drive_id: meta.drive_id, + bundle_id: meta.bundle_id, + bytes_written: data.len() as u64, + filemark_start, + filemark_end, + checksum: None, + success: true, + error: None, }) } } -fn phase1_unimplemented(op: &str) -> Status { - Status::unimplemented(format!( - "{op} is not implemented in phase-1 safe mode; no tape devices are accessed during unit-test runs" - )) +#[derive(Debug)] +pub struct SimulatorTapeBackend { + state: Mutex, +} + +impl SimulatorTapeBackend { + pub fn new(slot_count: u32, drive_count: u32) -> Self { + Self { + state: Mutex::new(SimulatorState::new(slot_count, drive_count)), + } + } + + pub fn from_config(config: &TapeConfig) -> Self { + let drive_count = config.scsi.devices.len().max(1) as u32; + Self::new(8, drive_count) + } + + pub fn insert_tape(&self, slot_id: &str, tape_id: &str) -> ServiceResult<()> { + let slot = parse_slot_id(slot_id)?; + let mut state = self.lock_state()?; + state + .library + .insert_tape(slot, TapeBarcode::new(tape_id)) + .map_err(vtl_status) + } + + fn lock_state(&self) -> ServiceResult> { + self.state + .lock() + .map_err(|_| Status::internal("simulator tape backend mutex poisoned")) + } +} + +#[derive(Debug)] +struct SimulatorState { + library: VirtualTapeLibrary, + acquired: Vec, + next_filemark: Vec, +} + +impl SimulatorState { + fn new(slot_count: u32, drive_count: u32) -> Self { + Self { + library: VirtualTapeLibrary::new(slot_count, drive_count), + acquired: vec![false; drive_count as usize], + next_filemark: vec![0; drive_count as usize], + } + } + + fn drive_endpoint(&self, drive_index: u32) -> ServiceResult { + let drive = self + .library + .drive(ElementAddress::drive(drive_index)) + .map_err(vtl_status)?; + let acquired = self + .acquired + .get(drive_index as usize) + .copied() + .unwrap_or(false); + Ok(common::DriveEndpoint { + drive_id: drive_id_string(drive_index), + device_path: format!("vtl://drive/{drive_index}"), + drive_type: "virtual-lto".to_string(), + status: if acquired { + DriveStatus::DriveInUse as i32 + } else { + DriveStatus::DriveIdle as i32 + }, + current_tape: drive + .loaded_barcode() + .map(|barcode| barcode.as_str().to_string()), + }) + } + + fn drive_index(&self, drive_id: &str) -> ServiceResult { + let address = parse_drive_id(drive_id)?; + self.library.drive(address).map_err(vtl_status)?; + Ok(address.index()) + } + + fn find_slot_with_tape(&self, tape_id: &str) -> ServiceResult { + self.library + .slots() + .iter() + .find(|slot| slot.barcode().map(|barcode| barcode.as_str()) == Some(tape_id)) + .map(|slot| slot.address()) + .ok_or_else(|| { + Status::not_found(format!("tape {tape_id} not found in simulator slots")) + }) + } + + fn find_empty_slot(&self) -> ServiceResult { + self.library + .slots() + .iter() + .find(|slot| slot.is_empty()) + .map(|slot| slot.address()) + .ok_or_else(|| Status::failed_precondition("no empty simulator slot is available")) + } +} + +impl TapeBackend for SimulatorTapeBackend { + fn list_drives(&self) -> ServiceResult> { + let state = self.lock_state()?; + (0..state.library.drives().len() as u32) + .map(|drive_index| state.drive_endpoint(drive_index)) + .collect() + } + + fn get_drive_status(&self, drive_id: &str) -> ServiceResult { + let state = self.lock_state()?; + let drive_index = state.drive_index(drive_id)?; + state.drive_endpoint(drive_index) + } + + fn acquire_drive( + &self, + preferred_drive_id: Option<&str>, + required_tape_id: Option<&str>, + ) -> ServiceResult { + let mut state = self.lock_state()?; + let drive_index = if let Some(preferred) = preferred_drive_id { + state.drive_index(preferred)? + } else { + state + .library + .drives() + .iter() + .find(|drive| { + !state.acquired[drive.address().index() as usize] + && required_tape_id.is_none_or(|tape_id| { + drive.loaded_barcode().map(|barcode| barcode.as_str()) == Some(tape_id) + }) + }) + .map(|drive| drive.address().index()) + .ok_or_else(|| Status::resource_exhausted("no matching simulator drive is free"))? + }; + + let index = drive_index as usize; + if state.acquired[index] { + return Err(Status::failed_precondition(format!( + "drive {} is already acquired", + drive_id_string(drive_index) + ))); + } + if let Some(tape_id) = required_tape_id { + let drive = state + .library + .drive(ElementAddress::drive(drive_index)) + .map_err(vtl_status)?; + if drive.loaded_barcode().map(|barcode| barcode.as_str()) != Some(tape_id) { + return Err(Status::failed_precondition(format!( + "drive {} does not contain required tape {tape_id}", + drive_id_string(drive_index) + ))); + } + } + + state.acquired[index] = true; + let drive = state + .library + .drive(ElementAddress::drive(drive_index)) + .map_err(vtl_status)?; + Ok(AcquireDriveResponse { + drive_id: drive_id_string(drive_index), + current_tape: drive + .loaded_barcode() + .map(|barcode| barcode.as_str().to_string()), + }) + } + + fn release_drive(&self, drive_id: &str) -> ServiceResult<()> { + let mut state = self.lock_state()?; + let drive_index = state.drive_index(drive_id)?; + state.acquired[drive_index as usize] = false; + Ok(()) + } + + fn load_tape(&self, tape_id: &str, drive_id: &str, slot_id: Option<&str>) -> ServiceResult<()> { + let drive = parse_drive_id(drive_id)?; + let mut state = self.lock_state()?; + state.library.drive(drive).map_err(vtl_status)?; + let slot = if let Some(slot_id) = slot_id { + parse_slot_id(slot_id)? + } else { + state.find_slot_with_tape(tape_id)? + }; + let slot_ref = state.library.slot(slot).map_err(vtl_status)?; + if slot_ref.barcode().map(|barcode| barcode.as_str()) != Some(tape_id) { + return Err(Status::failed_precondition(format!( + "slot {slot} does not contain tape {tape_id}" + ))); + } + state.library.load(slot, drive).map_err(vtl_status) + } + + fn unload_tape(&self, drive_id: &str, target_slot_id: Option<&str>) -> ServiceResult<()> { + let drive = parse_drive_id(drive_id)?; + let mut state = self.lock_state()?; + state.library.drive(drive).map_err(vtl_status)?; + let slot = if let Some(slot_id) = target_slot_id { + parse_slot_id(slot_id)? + } else { + state.find_empty_slot()? + }; + state.library.unload(drive, slot).map_err(vtl_status) + } + + fn rewind(&self, drive_id: &str) -> ServiceResult<()> { + let drive = parse_drive_id(drive_id)?; + let mut state = self.lock_state()?; + state.library.rewind(drive).map_err(vtl_status) + } + + fn seek_to_filemark(&self, drive_id: &str, filemark: u32) -> ServiceResult<()> { + let drive = parse_drive_id(drive_id)?; + let mut state = self.lock_state()?; + state.library.rewind(drive).map_err(vtl_status)?; + if filemark > 0 { + state + .library + .seek_filemark(drive, filemark) + .map_err(vtl_status)?; + } + Ok(()) + } + + fn get_tape_media_status(&self, drive_id: &str) -> ServiceResult { + let state = self.lock_state()?; + let drive_index = state.drive_index(drive_id)?; + let drive = state + .library + .drive(ElementAddress::drive(drive_index)) + .map_err(vtl_status)?; + let Some(tape) = drive.loaded_tape() else { + return Ok(TapeMediaStatus { + drive_id: drive_id_string(drive_index), + tape_id: None, + tape_status: TapeStatus::TapeOffline as i32, + capacity_bytes: SIMULATED_TAPE_CAPACITY_BYTES, + used_bytes: 0, + remaining_bytes: SIMULATED_TAPE_CAPACITY_BYTES, + current_position: 0, + current_filemark: 0, + is_write_protected: false, + }); + }; + + let used_bytes = tape.used_bytes(); + Ok(TapeMediaStatus { + drive_id: drive_id_string(drive_index), + tape_id: Some(tape.barcode().as_str().to_string()), + tape_status: TapeStatus::TapeOnline as i32, + capacity_bytes: SIMULATED_TAPE_CAPACITY_BYTES, + used_bytes, + remaining_bytes: SIMULATED_TAPE_CAPACITY_BYTES.saturating_sub(used_bytes), + current_position: tape.current_position(), + current_filemark: tape.current_filemark(), + is_write_protected: false, + }) + } + + fn inventory(&self) -> ServiceResult { + let state = self.lock_state()?; + let mut slots = Vec::new(); + for slot in state.library.slots() { + slots.push(SlotInfo { + slot_id: slot_id(slot.address().index()), + tape_id: slot.barcode().map(|barcode| barcode.as_str().to_string()), + is_drive: false, + drive_id: None, + is_import_export: false, + }); + } + for drive in state.library.drives() { + slots.push(SlotInfo { + slot_id: drive_id_string(drive.address().index()), + tape_id: drive + .loaded_barcode() + .map(|barcode| barcode.as_str().to_string()), + is_drive: true, + drive_id: Some(drive_id_string(drive.address().index())), + is_import_export: false, + }); + } + Ok(InventoryResponse { slots }) + } + + fn write_bundle(&self, drive_id: &str, data: &[u8]) -> ServiceResult<(u32, u32)> { + let drive = parse_drive_id(drive_id)?; + let mut state = self.lock_state()?; + state.library.drive(drive).map_err(vtl_status)?; + let drive_index = drive.index() as usize; + let filemark_start = state.next_filemark[drive_index]; + state.library.write(drive, data).map_err(vtl_status)?; + state.library.write_filemark(drive).map_err(vtl_status)?; + state.next_filemark[drive_index] += 1; + Ok((filemark_start, state.next_filemark[drive_index])) + } + + fn read_bundle(&self, drive_id: &str, filemark: u32, length: u64) -> ServiceResult> { + let drive = parse_drive_id(drive_id)?; + let mut state = self.lock_state()?; + state.library.rewind(drive).map_err(vtl_status)?; + if filemark > 0 { + state + .library + .seek_filemark(drive, filemark) + .map_err(vtl_status)?; + } + let max_len = if length == 0 { + usize::MAX + } else { + length.min(usize::MAX as u64) as usize + }; + state.library.read(drive, max_len).map_err(vtl_status) + } } #[tonic::async_trait] impl TapeService for TapeServiceImpl { async fn write_bundle( &self, - _req: Request>, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.write_bundle")) + req: Request>, + ) -> ServiceResult> { + self.write_bundle_from_messages(req.into_inner()) + .await + .map(Response::new) } - type ReadBundleStream = - tokio_stream::wrappers::ReceiverStream>; + type ReadBundleStream = ReceiverStream>; async fn read_bundle( &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.read_bundle")) + req: Request, + ) -> ServiceResult> { + let req = req.into_inner(); + let filemark = match req.location { + Some(Location::Filemark(filemark)) => filemark, + Some(Location::BlockOffset(_)) => { + return Err(Status::unimplemented( + "phase-1 simulator supports filemark-based reads; block offset reads require live SCSI/tape integration", + )) + } + None => return Err(Status::invalid_argument("read_bundle location is required")), + }; + let data = self + .backend + .read_bundle(&req.drive_id, filemark, req.length)?; + let (tx, rx) = mpsc::channel(2); + tx.send(Ok(ReadBundleResponse { + payload: Some(ReadPayload::Meta(ReadBundleMeta { + total_size: data.len() as u64, + checksum: None, + })), + })) + .await + .map_err(|_| Status::cancelled("read_bundle receiver dropped before metadata send"))?; + if !data.is_empty() { + tx.send(Ok(ReadBundleResponse { + payload: Some(ReadPayload::Data(data)), + })) + .await + .map_err(|_| Status::cancelled("read_bundle receiver dropped before data send"))?; + } + Ok(Response::new(ReceiverStream::new(rx))) } - async fn list_drives( - &self, - _req: Request<()>, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.list_drives")) + async fn list_drives(&self, _req: Request<()>) -> ServiceResult> { + Ok(Response::new(ListDrivesResponse { + drives: self.backend.list_drives()?, + })) } async fn get_drive_status( &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.get_drive_status")) + req: Request, + ) -> ServiceResult> { + Ok(Response::new( + self.backend.get_drive_status(&req.into_inner().drive_id)?, + )) } async fn acquire_drive( &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.acquire_drive")) + req: Request, + ) -> ServiceResult> { + let req = req.into_inner(); + Ok(Response::new(self.backend.acquire_drive( + req.preferred_drive_id.as_deref(), + req.required_tape_id.as_deref(), + )?)) } async fn release_drive( &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.release_drive")) + req: Request, + ) -> ServiceResult> { + self.backend.release_drive(&req.into_inner().drive_id)?; + Ok(Response::new(())) } - async fn load_tape( - &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.load_tape")) + async fn load_tape(&self, req: Request) -> ServiceResult> { + let req = req.into_inner(); + self.backend + .load_tape(&req.tape_id, &req.drive_id, req.slot_id.as_deref())?; + Ok(Response::new(())) } - async fn unload_tape( - &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.unload_tape")) + async fn unload_tape(&self, req: Request) -> ServiceResult> { + let req = req.into_inner(); + self.backend + .unload_tape(&req.drive_id, req.target_slot_id.as_deref())?; + Ok(Response::new(())) } - async fn rewind( - &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.rewind")) + async fn rewind(&self, req: Request) -> ServiceResult> { + self.backend.rewind(&req.into_inner().drive_id)?; + Ok(Response::new(())) } async fn seek_to_filemark( &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.seek_to_filemark")) + req: Request, + ) -> ServiceResult> { + let req = req.into_inner(); + self.backend.seek_to_filemark(&req.drive_id, req.filemark)?; + Ok(Response::new(())) } async fn get_tape_media_status( &self, - _req: Request, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.get_tape_media_status")) + req: Request, + ) -> ServiceResult> { + Ok(Response::new( + self.backend + .get_tape_media_status(&req.into_inner().drive_id)?, + )) } - async fn inventory( - &self, - _req: Request<()>, - ) -> std::result::Result, Status> { - Err(phase1_unimplemented("tape.inventory")) + async fn inventory(&self, _req: Request<()>) -> ServiceResult> { + Ok(Response::new(self.backend.inventory()?)) + } +} + +fn parse_slot_id(slot_id: &str) -> ServiceResult { + let raw = slot_id.strip_prefix("slot-").unwrap_or(slot_id); + let index = raw + .parse::() + .map_err(|_| Status::invalid_argument(format!("invalid slot id: {slot_id}")))?; + if index == 0 { + return Err(Status::invalid_argument("slot ids are 1-based")); + } + Ok(ElementAddress::slot(index)) +} + +fn parse_drive_id(drive_id: &str) -> ServiceResult { + let raw = drive_id.strip_prefix("drive-").unwrap_or(drive_id); + let index = raw + .parse::() + .map_err(|_| Status::invalid_argument(format!("invalid drive id: {drive_id}")))?; + Ok(ElementAddress::drive(index)) +} + +fn slot_id(index: u32) -> String { + format!("slot-{index}") +} + +fn drive_id_string(index: u32) -> String { + format!("drive-{index}") +} + +fn vtl_status(error: coldstore_vtl::VtlError) -> Status { + match error { + coldstore_vtl::VtlError::SlotEmpty(_) + | coldstore_vtl::VtlError::SlotOccupied(_) + | coldstore_vtl::VtlError::DriveEmpty(_) + | coldstore_vtl::VtlError::DriveOccupied(_) + | coldstore_vtl::VtlError::FilemarkNotFound => { + Status::failed_precondition(error.to_string()) + } + coldstore_vtl::VtlError::SlotOutOfRange(_) + | coldstore_vtl::VtlError::DriveOutOfRange(_) + | coldstore_vtl::VtlError::WrongElement { .. } + | coldstore_vtl::VtlError::InvalidLsscsiLine(_) + | coldstore_vtl::VtlError::InvalidElementAddress(_) => { + Status::invalid_argument(error.to_string()) + } + coldstore_vtl::VtlError::CommandFailed { .. } => Status::unavailable(error.to_string()), + coldstore_vtl::VtlError::Io(_) => Status::unavailable(error.to_string()), } } diff --git a/crates/tape/tests/simulator_service.rs b/crates/tape/tests/simulator_service.rs new file mode 100644 index 0000000..dcedc86 --- /dev/null +++ b/crates/tape/tests/simulator_service.rs @@ -0,0 +1,179 @@ +use coldstore_common::config::TapeConfig; +use coldstore_proto::common::{DriveStatus, TapeStatus}; +use coldstore_proto::tape::read_bundle_request::Location; +use coldstore_proto::tape::read_bundle_response::Payload as ReadPayload; +use coldstore_proto::tape::tape_service_server::TapeService; +use coldstore_proto::tape::write_bundle_request::Payload as WritePayload; +use coldstore_proto::tape::{ + LoadTapeRequest, ReadBundleRequest, RewindRequest, SeekToFilemarkRequest, UnloadTapeRequest, + WriteBundleMeta, WriteBundleRequest, +}; +use coldstore_tape::service::{SimulatorTapeBackend, TapeServiceImpl}; +use tokio_stream::StreamExt; +use tonic::Request; + +#[tokio::test] +async fn simulator_backend_exposes_inventory_and_drive_operations_through_service() { + let backend = SimulatorTapeBackend::new(3, 1); + backend.insert_tape("slot-1", "TAPE0001L9").unwrap(); + let service = TapeServiceImpl::new_with_backend(TapeConfig::default(), backend); + + let drives = service + .list_drives(Request::new(())) + .await + .unwrap() + .into_inner() + .drives; + assert_eq!(drives.len(), 1); + assert_eq!(drives[0].drive_id, "drive-0"); + assert_eq!(drives[0].status, DriveStatus::DriveIdle as i32); + assert_eq!(drives[0].current_tape.as_deref(), None); + + let inventory = service + .inventory(Request::new(())) + .await + .unwrap() + .into_inner(); + let slot_1 = inventory + .slots + .iter() + .find(|slot| slot.slot_id == "slot-1") + .unwrap(); + assert!(!slot_1.is_drive); + assert_eq!(slot_1.tape_id.as_deref(), Some("TAPE0001L9")); + + service + .load_tape(Request::new(LoadTapeRequest { + tape_id: "TAPE0001L9".to_string(), + drive_id: "drive-0".to_string(), + slot_id: None, + })) + .await + .unwrap(); + + let status = service + .get_drive_status(Request::new(coldstore_proto::tape::GetDriveStatusRequest { + drive_id: "drive-0".to_string(), + })) + .await + .unwrap() + .into_inner(); + assert_eq!(status.current_tape.as_deref(), Some("TAPE0001L9")); + + let media = service + .get_tape_media_status(Request::new( + coldstore_proto::tape::GetTapeMediaStatusRequest { + drive_id: "drive-0".to_string(), + }, + )) + .await + .unwrap() + .into_inner(); + assert_eq!(media.tape_id.as_deref(), Some("TAPE0001L9")); + assert_eq!(media.tape_status, TapeStatus::TapeOnline as i32); + + service + .unload_tape(Request::new(UnloadTapeRequest { + drive_id: "drive-0".to_string(), + target_slot_id: Some("slot-2".to_string()), + })) + .await + .unwrap(); + + let inventory = service + .inventory(Request::new(())) + .await + .unwrap() + .into_inner(); + let slot_2 = inventory + .slots + .iter() + .find(|slot| slot.slot_id == "slot-2") + .unwrap(); + assert_eq!(slot_2.tape_id.as_deref(), Some("TAPE0001L9")); +} + +#[tokio::test] +async fn simulator_service_writes_filemark_delimited_bundles_and_reads_them_back() { + let backend = SimulatorTapeBackend::new(2, 1); + backend.insert_tape("slot-1", "TAPE0002L9").unwrap(); + let service = TapeServiceImpl::new_with_backend(TapeConfig::default(), backend); + + service + .load_tape(Request::new(LoadTapeRequest { + tape_id: "TAPE0002L9".to_string(), + drive_id: "drive-0".to_string(), + slot_id: Some("slot-1".to_string()), + })) + .await + .unwrap(); + + let response = service + .write_bundle_from_messages(tokio_stream::iter(vec![ + Ok(WriteBundleRequest { + payload: Some(WritePayload::Meta(WriteBundleMeta { + drive_id: "drive-0".to_string(), + bundle_id: "bundle-a".to_string(), + total_size: 6, + object_count: 1, + block_size: 262_144, + })), + }), + Ok(WriteBundleRequest { + payload: Some(WritePayload::Data(b"abc".to_vec())), + }), + Ok(WriteBundleRequest { + payload: Some(WritePayload::Data(b"def".to_vec())), + }), + ])) + .await + .unwrap(); + assert!(response.success); + assert_eq!(response.bundle_id, "bundle-a"); + assert_eq!(response.bytes_written, 6); + assert_eq!(response.filemark_start, 0); + assert_eq!(response.filemark_end, 1); + + let media = service + .get_tape_media_status(Request::new( + coldstore_proto::tape::GetTapeMediaStatusRequest { + drive_id: "drive-0".to_string(), + }, + )) + .await + .unwrap() + .into_inner(); + assert_eq!(media.used_bytes, 6); + assert_eq!(media.current_filemark, 1); + + service + .rewind(Request::new(RewindRequest { + drive_id: "drive-0".to_string(), + })) + .await + .unwrap(); + + let mut stream = service + .read_bundle(Request::new(ReadBundleRequest { + drive_id: "drive-0".to_string(), + location: Some(Location::Filemark(0)), + length: 6, + })) + .await + .unwrap() + .into_inner(); + + let meta = stream.next().await.unwrap().unwrap().payload.unwrap(); + assert!(matches!(meta, ReadPayload::Meta(_))); + let data = stream.next().await.unwrap().unwrap().payload.unwrap(); + assert_eq!(data, ReadPayload::Data(b"abcdef".to_vec())); + assert!(stream.next().await.is_none()); + + service + .seek_to_filemark(Request::new(SeekToFilemarkRequest { + drive_id: "drive-0".to_string(), + filemark: 1, + })) + .await + .unwrap(); +} diff --git a/crates/vtl/Cargo.toml b/crates/vtl/Cargo.toml new file mode 100644 index 0000000..68ec36b --- /dev/null +++ b/crates/vtl/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "coldstore-vtl" +description = "ColdStore virtual tape library harness and mhVTL command wrappers" +version.workspace = true +edition.workspace = true +license.workspace = true + +[dependencies] +thiserror = { workspace = true } + +[dev-dependencies] diff --git a/crates/vtl/README.md b/crates/vtl/README.md new file mode 100644 index 0000000..8a009f6 --- /dev/null +++ b/crates/vtl/README.md @@ -0,0 +1,51 @@ +# coldstore-vtl + +ColdStore 的虚拟磁带库 harness。该 crate 的目标不是在开发机上自动安装或启动 mhVTL，而是先把后续 TapeService 闭环开发需要的边界固化下来： + +- 安全单测：不执行系统命令、不加载内核模块、不创建 `/dev/st*` / `/dev/sg*` / `/dev/sch*`。 +- 命令封装：稳定暴露 `lsscsi`、`mtx`、`mt`、`sg3_utils`、`dd`、`vtlcmd` 的命令构造。 +- 真实后端预留：在独立 VM/专用测试机安装 mhVTL 后，用 `SystemCommandRunner` 执行相同命令。 +- 内存模拟器：提供 medium changer + tape drive 的 slot/barcode/load/unload/rewind/filemark/read/write 行为，供 phase-1/phase-2 单测使用。 + +## 模块分层 + +| 模块 | 职责 | 是否触碰宿主 | +|---|---|---| +| `model` | `TapeBarcode`、`ElementAddress`、`VirtualTape`、filemark/cursor 模型 | 否 | +| `interface` | `MediumChanger`、`TapeDrive`、`TapeInventory` trait | 否 | +| `simulator` | 纯内存 VTL，模拟 slot、drive、barcode、load/unload、rewind、filemark、read/write | 否 | +| `discover` | 解析 `lsscsi -g` 输出为 `ScsiInventory` | 否 | +| `command` | `CommandSpec`、`CommandRunner`、`SystemCommandRunner`、`RecordedCommandRunner` | 只有 `SystemCommandRunner` 会执行 | +| `mhvtl` | mhVTL/live 工具链命令构造和可选执行入口 | 构造命令不触碰；执行需显式 runner | + +## 安全验证 + +```bash +cargo test -p coldstore-vtl --lib --tests +``` + +这只运行单测，不安装系统工具，不访问真实设备。 + +## 后续 live mhVTL 使用边界 + +live 环境必须放在专用 VM 或测试机中，不建议直接在当前开发宿主机执行： + +1. 执行前人工阅读 `scripts/setup-mhvtl-env.sh`。 +2. 只在专用 VM 中运行： + ```bash + scripts/setup-mhvtl-env.sh --execute --start-services + ``` +3. 验证设备发现： + ```bash + lsscsi -g + ``` +4. 用 `coldstore-vtl::mhvtl::MhvtlToolchain` 生成命令，再由上层测试 harness 决定是否执行。 + +## 当前已覆盖能力 + +- `lsscsi -g` 样例解析：识别 `mediumx` / `tape`，抽取 HCTL、vendor、product、revision、`/dev/schX`、`/dev/stX`、`/dev/nstX`、`/dev/sgX`。 +- `mtx` 命令封装：status、transfer、load、unload。 +- `mt` 命令封装：status、rewind、offline、weof、fsf。 +- `sg3_utils` 命令封装：sg_inq、sg_turs、sg_logs、sg_modes。 +- `dd` 命令封装：向 tape device 写入/读取数据。 +- 内存 VTL：slot 插带、load/unload、drive 读写、rewind、filemark seek。 diff --git a/crates/vtl/src/command.rs b/crates/vtl/src/command.rs new file mode 100644 index 0000000..e02c652 --- /dev/null +++ b/crates/vtl/src/command.rs @@ -0,0 +1,146 @@ +use crate::error::{Result, VtlError}; +use std::cell::RefCell; +use std::collections::VecDeque; +use std::fmt; +use std::process::Command; + +#[derive(Debug, Clone, Eq, PartialEq, Hash)] +pub struct CommandSpec { + program: String, + args: Vec, +} + +impl CommandSpec { + pub fn new(program: impl Into) -> Self { + Self { + program: program.into(), + args: Vec::new(), + } + } + + pub fn arg(mut self, arg: impl Into) -> Self { + self.args.push(arg.into()); + self + } + + pub fn args(mut self, args: I) -> Self + where + I: IntoIterator, + S: Into, + { + self.args.extend(args.into_iter().map(Into::into)); + self + } + + pub fn program(&self) -> &str { + &self.program + } + + pub fn arguments(&self) -> &[String] { + &self.args + } + + pub fn argv(&self) -> Vec<&str> { + std::iter::once(self.program.as_str()) + .chain(self.args.iter().map(String::as_str)) + .collect() + } +} + +impl fmt::Display for CommandSpec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", shell_quote(&self.program))?; + for arg in &self.args { + write!(f, " {}", shell_quote(arg))?; + } + Ok(()) + } +} + +#[derive(Debug, Clone, Eq, PartialEq, Default)] +pub struct CommandOutput { + pub status: i32, + pub stdout: String, + pub stderr: String, +} + +impl CommandOutput { + pub fn success(stdout: impl Into) -> Self { + Self { + status: 0, + stdout: stdout.into(), + stderr: String::new(), + } + } +} + +pub trait CommandRunner { + fn run(&self, command: &CommandSpec) -> Result; + + fn run_checked(&self, command: &CommandSpec) -> Result { + let output = self.run(command)?; + if output.status == 0 { + Ok(output) + } else { + Err(VtlError::CommandFailed { + command: command.clone(), + status: output.status, + stderr: output.stderr, + }) + } + } +} + +#[derive(Debug, Default, Copy, Clone)] +pub struct SystemCommandRunner; + +impl CommandRunner for SystemCommandRunner { + fn run(&self, command: &CommandSpec) -> Result { + let output = Command::new(command.program()) + .args(command.arguments()) + .output()?; + Ok(CommandOutput { + status: output.status.code().unwrap_or(-1), + stdout: String::from_utf8_lossy(&output.stdout).into_owned(), + stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + }) + } +} + +#[derive(Debug, Default)] +pub struct RecordedCommandRunner { + commands: RefCell>, + responses: RefCell>, +} + +impl RecordedCommandRunner { + pub fn push_response(&self, response: CommandOutput) { + self.responses.borrow_mut().push_back(response); + } + + pub fn push_stdout(&self, stdout: impl Into) { + self.push_response(CommandOutput::success(stdout)); + } + + pub fn commands(&self) -> Vec { + self.commands.borrow().clone() + } +} + +impl CommandRunner for RecordedCommandRunner { + fn run(&self, command: &CommandSpec) -> Result { + self.commands.borrow_mut().push(command.clone()); + Ok(self.responses.borrow_mut().pop_front().unwrap_or_default()) + } +} + +fn shell_quote(value: &str) -> String { + if value + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '/' | '_' | '-' | '.' | ':' | '=')) + { + value.to_string() + } else { + format!("'{}'", value.replace("'", "'\\''")) + } +} diff --git a/crates/vtl/src/discover.rs b/crates/vtl/src/discover.rs new file mode 100644 index 0000000..7bd8309 --- /dev/null +++ b/crates/vtl/src/discover.rs @@ -0,0 +1,144 @@ +use crate::error::{Result, VtlError}; + +#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)] +pub enum ScsiDeviceKind { + TapeDrive, + MediumChanger, + Disk, + CdDvd, + Other, +} + +impl ScsiDeviceKind { + fn from_lsscsi(value: &str) -> Self { + match value { + "tape" => Self::TapeDrive, + "mediumx" => Self::MediumChanger, + "disk" => Self::Disk, + "cd/dvd" => Self::CdDvd, + _ => Self::Other, + } + } +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct ScsiDevice { + pub hctl: String, + pub kind: ScsiDeviceKind, + pub vendor: String, + pub product: String, + pub revision: String, + pub primary_device: Option, + pub sg_device: Option, + pub non_rewinding_device: Option, +} + +#[derive(Debug, Clone, Eq, PartialEq, Default)] +pub struct ScsiInventory { + devices: Vec, +} + +impl ScsiInventory { + pub fn new(devices: Vec) -> Self { + Self { devices } + } + + pub fn devices(&self) -> &[ScsiDevice] { + &self.devices + } + + pub fn tape_drives(&self) -> Vec<&ScsiDevice> { + self.devices + .iter() + .filter(|device| device.kind == ScsiDeviceKind::TapeDrive) + .collect() + } + + pub fn medium_changers(&self) -> Vec<&ScsiDevice> { + self.devices + .iter() + .filter(|device| device.kind == ScsiDeviceKind::MediumChanger) + .collect() + } +} + +pub fn parse_lsscsi(output: &str) -> Result { + let mut devices = Vec::new(); + for line in output + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + { + devices.push(parse_lsscsi_line(line)?); + } + Ok(ScsiInventory::new(devices)) +} + +fn parse_lsscsi_line(line: &str) -> Result { + let Some(close) = line.find(']') else { + return Err(VtlError::InvalidLsscsiLine(line.into())); + }; + let hctl = line + .strip_prefix('[') + .and_then(|rest| rest.get(..close - 1)) + .ok_or_else(|| VtlError::InvalidLsscsiLine(line.into()))? + .to_string(); + + let tokens: Vec<&str> = line[close + 1..].split_whitespace().collect(); + if tokens.len() < 4 { + return Err(VtlError::InvalidLsscsiLine(line.into())); + } + + let kind = ScsiDeviceKind::from_lsscsi(tokens[0]); + let vendor = tokens[1].to_string(); + let product = tokens[2].to_string(); + let revision = tokens[3].to_string(); + let primary_device = tokens + .get(4) + .filter(|value| value.starts_with("/dev/")) + .map(|value| (*value).to_string()); + let sg_device = tokens + .get(5) + .filter(|value| value.starts_with("/dev/sg")) + .map(|value| (*value).to_string()) + .or_else(|| { + tokens + .iter() + .find(|value| value.starts_with("/dev/sg")) + .map(|value| (*value).to_string()) + }); + let non_rewinding_device = primary_device + .as_deref() + .and_then(non_rewinding_tape_device); + + Ok(ScsiDevice { + hctl, + kind, + vendor, + product, + revision, + primary_device, + sg_device, + non_rewinding_device, + }) +} + +fn non_rewinding_tape_device(path: &str) -> Option { + let suffix = path.strip_prefix("/dev/st")?; + if suffix.chars().all(|c| c.is_ascii_digit()) { + Some(format!("/dev/nst{suffix}")) + } else { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_non_lsscsi_lines() { + let err = parse_lsscsi_line("not a device").expect_err("invalid line"); + assert!(matches!(err, VtlError::InvalidLsscsiLine(_))); + } +} diff --git a/crates/vtl/src/error.rs b/crates/vtl/src/error.rs new file mode 100644 index 0000000..9925c60 --- /dev/null +++ b/crates/vtl/src/error.rs @@ -0,0 +1,50 @@ +use crate::command::CommandSpec; +use thiserror::Error; + +pub type Result = std::result::Result; + +#[derive(Debug, Error)] +pub enum VtlError { + #[error("invalid SCSI/lsscsi line: {0}")] + InvalidLsscsiLine(String), + + #[error("invalid element address: {0}")] + InvalidElementAddress(String), + + #[error("slot {0} does not exist")] + SlotOutOfRange(u32), + + #[error("drive {0} does not exist")] + DriveOutOfRange(u32), + + #[error("slot {0} is empty")] + SlotEmpty(u32), + + #[error("slot {0} is already occupied")] + SlotOccupied(u32), + + #[error("drive {0} is empty")] + DriveEmpty(u32), + + #[error("drive {0} is already occupied")] + DriveOccupied(u32), + + #[error("expected {expected} element, got {actual}")] + WrongElement { + expected: &'static str, + actual: String, + }, + + #[error("filemark not found before end of tape")] + FilemarkNotFound, + + #[error("command failed: {command}; status={status}; stderr={stderr}")] + CommandFailed { + command: CommandSpec, + status: i32, + stderr: String, + }, + + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), +} diff --git a/crates/vtl/src/interface.rs b/crates/vtl/src/interface.rs new file mode 100644 index 0000000..df12a67 --- /dev/null +++ b/crates/vtl/src/interface.rs @@ -0,0 +1,20 @@ +use crate::error::Result; +use crate::model::{ElementAddress, TapeBarcode}; + +pub trait MediumChanger { + fn move_medium(&mut self, from: ElementAddress, to: ElementAddress) -> Result<()>; + fn load(&mut self, slot: ElementAddress, drive: ElementAddress) -> Result<()>; + fn unload(&mut self, drive: ElementAddress, slot: ElementAddress) -> Result<()>; +} + +pub trait TapeDrive { + fn rewind(&mut self, drive: ElementAddress) -> Result<()>; + fn seek_filemark(&mut self, drive: ElementAddress, count: u32) -> Result<()>; + fn write_filemark(&mut self, drive: ElementAddress) -> Result<()>; + fn write(&mut self, drive: ElementAddress, data: &[u8]) -> Result<()>; + fn read(&mut self, drive: ElementAddress, max_len: usize) -> Result>; +} + +pub trait TapeInventory { + fn insert_tape(&mut self, slot: ElementAddress, barcode: TapeBarcode) -> Result<()>; +} diff --git a/crates/vtl/src/lib.rs b/crates/vtl/src/lib.rs new file mode 100644 index 0000000..3051ae4 --- /dev/null +++ b/crates/vtl/src/lib.rs @@ -0,0 +1,22 @@ +//! ColdStore virtual tape library harness and mhVTL command wrappers. +//! +//! The crate is split into safe, unit-testable pieces: +//! - [`model`] contains device/tape/changer value types. +//! - [`discover`] parses `lsscsi -g` output without touching the host. +//! - [`command`] describes and executes external commands behind a small trait. +//! - [`mhvtl`] exposes stable `lsscsi`/`mtx`/`mt`/`sg3_utils` command builders. +//! - [`simulator`] provides an in-memory VTL for unit tests and phase-1 logic. +//! +//! Live mhVTL usage is intentionally explicit: callers must provide a +//! [`command::SystemCommandRunner`] or another runner that actually executes +//! host commands. + +pub mod command; +pub mod discover; +pub mod error; +pub mod interface; +pub mod mhvtl; +pub mod model; +pub mod simulator; + +pub use error::{Result, VtlError}; diff --git a/crates/vtl/src/mhvtl.rs b/crates/vtl/src/mhvtl.rs new file mode 100644 index 0000000..8727220 --- /dev/null +++ b/crates/vtl/src/mhvtl.rs @@ -0,0 +1,208 @@ +use crate::command::{CommandOutput, CommandRunner, CommandSpec}; +use crate::discover::{parse_lsscsi, ScsiInventory}; +use crate::error::Result; +use crate::model::ElementAddress; +use std::path::Path; + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct ToolPaths { + pub lsscsi: String, + pub mtx: String, + pub mt: String, + pub dd: String, + pub sg_inq: String, + pub sg_turs: String, + pub sg_logs: String, + pub sg_modes: String, + pub vtlcmd: String, +} + +impl Default for ToolPaths { + fn default() -> Self { + Self { + lsscsi: "lsscsi".into(), + mtx: "mtx".into(), + mt: "mt".into(), + dd: "dd".into(), + sg_inq: "sg_inq".into(), + sg_turs: "sg_turs".into(), + sg_logs: "sg_logs".into(), + sg_modes: "sg_modes".into(), + vtlcmd: "vtlcmd".into(), + } + } +} + +#[derive(Debug, Clone, Eq, PartialEq, Default)] +pub struct MhvtlToolchain { + paths: ToolPaths, +} + +impl MhvtlToolchain { + pub fn new(paths: ToolPaths) -> Self { + Self { paths } + } + + pub fn paths(&self) -> &ToolPaths { + &self.paths + } + + pub fn discover_command(&self) -> CommandSpec { + CommandSpec::new(self.paths.lsscsi.clone()).arg("-g") + } + + pub fn library_status_command(&self, changer_sg: &str) -> CommandSpec { + CommandSpec::new(self.paths.mtx.clone()).args(["-f", changer_sg, "status"]) + } + + pub fn move_medium_command( + &self, + changer_sg: &str, + from: ElementAddress, + to: ElementAddress, + ) -> CommandSpec { + CommandSpec::new(self.paths.mtx.clone()).args([ + "-f".to_string(), + changer_sg.to_string(), + "transfer".to_string(), + from.mtx_address().to_string(), + to.mtx_address().to_string(), + ]) + } + + pub fn load_command( + &self, + changer_sg: &str, + slot: ElementAddress, + drive: ElementAddress, + ) -> CommandSpec { + CommandSpec::new(self.paths.mtx.clone()).args([ + "-f".to_string(), + changer_sg.to_string(), + "load".to_string(), + slot.mtx_address().to_string(), + drive.mtx_address().to_string(), + ]) + } + + pub fn unload_command( + &self, + changer_sg: &str, + slot: ElementAddress, + drive: ElementAddress, + ) -> CommandSpec { + CommandSpec::new(self.paths.mtx.clone()).args([ + "-f".to_string(), + changer_sg.to_string(), + "unload".to_string(), + slot.mtx_address().to_string(), + drive.mtx_address().to_string(), + ]) + } + + pub fn tape_status_command(&self, tape_device: &str) -> CommandSpec { + CommandSpec::new(self.paths.mt.clone()).args(["-f", tape_device, "status"]) + } + + pub fn rewind_command(&self, tape_device: &str) -> CommandSpec { + CommandSpec::new(self.paths.mt.clone()).args(["-f", tape_device, "rewind"]) + } + + pub fn offline_command(&self, tape_device: &str) -> CommandSpec { + CommandSpec::new(self.paths.mt.clone()).args(["-f", tape_device, "offline"]) + } + + pub fn write_filemark_command(&self, tape_device: &str, count: u32) -> CommandSpec { + CommandSpec::new(self.paths.mt.clone()).args([ + "-f".to_string(), + tape_device.to_string(), + "weof".to_string(), + count.to_string(), + ]) + } + + pub fn seek_filemark_command(&self, tape_device: &str, count: u32) -> CommandSpec { + CommandSpec::new(self.paths.mt.clone()).args([ + "-f".to_string(), + tape_device.to_string(), + "fsf".to_string(), + count.to_string(), + ]) + } + + pub fn write_tape_command( + &self, + input_path: impl AsRef, + tape_device: &str, + block_size: u32, + ) -> CommandSpec { + CommandSpec::new(self.paths.dd.clone()).args([ + format!("if={}", input_path.as_ref().display()), + format!("of={tape_device}"), + format!("bs={block_size}"), + "status=none".to_string(), + ]) + } + + pub fn read_tape_command( + &self, + tape_device: &str, + output_path: impl AsRef, + block_size: u32, + count: u32, + ) -> CommandSpec { + CommandSpec::new(self.paths.dd.clone()).args([ + format!("if={tape_device}"), + format!("of={}", output_path.as_ref().display()), + format!("bs={block_size}"), + format!("count={count}"), + "iflag=fullblock".to_string(), + "status=none".to_string(), + ]) + } + + pub fn inquiry_command(&self, sg_device: &str) -> CommandSpec { + CommandSpec::new(self.paths.sg_inq.clone()).arg(sg_device) + } + + pub fn test_unit_ready_command(&self, sg_device: &str) -> CommandSpec { + CommandSpec::new(self.paths.sg_turs.clone()).arg(sg_device) + } + + pub fn sg_logs_command(&self, sg_device: &str) -> CommandSpec { + CommandSpec::new(self.paths.sg_logs.clone()).arg(sg_device) + } + + pub fn sg_modes_command(&self, sg_device: &str) -> CommandSpec { + CommandSpec::new(self.paths.sg_modes.clone()).arg(sg_device) + } + + pub fn vtlcmd_command(&self, args: I) -> CommandSpec + where + I: IntoIterator, + S: Into, + { + CommandSpec::new(self.paths.vtlcmd.clone()).args(args) + } + + pub fn discover(&self, runner: &dyn CommandRunner) -> Result { + let output = runner.run_checked(&self.discover_command())?; + parse_lsscsi(&output.stdout) + } + + pub fn run_library_status( + &self, + runner: &dyn CommandRunner, + changer_sg: &str, + ) -> Result { + runner.run_checked(&self.library_status_command(changer_sg)) + } + + pub fn run_test_unit_ready( + &self, + runner: &dyn CommandRunner, + sg_device: &str, + ) -> Result { + runner.run_checked(&self.test_unit_ready_command(sg_device)) + } +} diff --git a/crates/vtl/src/model.rs b/crates/vtl/src/model.rs new file mode 100644 index 0000000..1ab3767 --- /dev/null +++ b/crates/vtl/src/model.rs @@ -0,0 +1,223 @@ +use std::fmt; + +#[derive(Debug, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)] +pub struct TapeBarcode(String); + +impl TapeBarcode { + pub fn new(value: impl Into) -> Self { + Self(value.into()) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for TapeBarcode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)] +pub enum ElementKind { + Slot, + Drive, + ImportExport, +} + +impl fmt::Display for ElementKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ElementKind::Slot => f.write_str("slot"), + ElementKind::Drive => f.write_str("drive"), + ElementKind::ImportExport => f.write_str("import-export"), + } + } +} + +#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)] +pub struct ElementAddress { + kind: ElementKind, + index: u32, +} + +impl ElementAddress { + pub fn slot(index: u32) -> Self { + Self { + kind: ElementKind::Slot, + index, + } + } + + pub fn drive(index: u32) -> Self { + Self { + kind: ElementKind::Drive, + index, + } + } + + pub fn import_export(index: u32) -> Self { + Self { + kind: ElementKind::ImportExport, + index, + } + } + + pub fn kind(self) -> ElementKind { + self.kind + } + + pub fn index(self) -> u32 { + self.index + } + + /// Numeric element value used by `mtx` style commands. + /// + /// `mtx load/unload` uses slot number + drive index; `mtx transfer` accepts + /// numeric element addresses. We keep this value explicit so live wrappers + /// can be reviewed before they touch a host. + pub fn mtx_address(self) -> u32 { + self.index + } +} + +impl fmt::Display for ElementAddress { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}:{}", self.kind, self.index) + } +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub enum TapeRecord { + Data(Vec), + Filemark, +} + +#[derive(Debug, Clone, Eq, PartialEq, Default)] +pub struct TapeCursor { + pub record_index: usize, + pub byte_offset: usize, +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct VirtualTape { + barcode: TapeBarcode, + records: Vec, + cursor: TapeCursor, +} + +impl VirtualTape { + pub fn new(barcode: TapeBarcode) -> Self { + Self { + barcode, + records: Vec::new(), + cursor: TapeCursor::default(), + } + } + + pub fn barcode(&self) -> &TapeBarcode { + &self.barcode + } + + pub fn records(&self) -> &[TapeRecord] { + &self.records + } + + pub fn cursor(&self) -> &TapeCursor { + &self.cursor + } + + pub fn used_bytes(&self) -> u64 { + self.records + .iter() + .map(|record| match record { + TapeRecord::Data(data) => data.len() as u64, + TapeRecord::Filemark => 0, + }) + .sum() + } + + pub fn current_position(&self) -> u64 { + let completed_bytes: u64 = self.records[..self.cursor.record_index.min(self.records.len())] + .iter() + .map(|record| match record { + TapeRecord::Data(data) => data.len() as u64, + TapeRecord::Filemark => 0, + }) + .sum(); + completed_bytes + self.cursor.byte_offset as u64 + } + + pub fn current_filemark(&self) -> u32 { + self.records[..self.cursor.record_index.min(self.records.len())] + .iter() + .filter(|record| matches!(record, TapeRecord::Filemark)) + .count() as u32 + } + + pub(crate) fn rewind(&mut self) { + self.cursor = TapeCursor::default(); + } + + pub(crate) fn append_data(&mut self, data: &[u8]) { + if self.cursor.record_index < self.records.len() { + self.records.truncate(self.cursor.record_index); + } + self.records.push(TapeRecord::Data(data.to_vec())); + self.cursor.record_index = self.records.len(); + self.cursor.byte_offset = 0; + } + + pub(crate) fn append_filemark(&mut self) { + if self.cursor.record_index < self.records.len() { + self.records.truncate(self.cursor.record_index); + } + self.records.push(TapeRecord::Filemark); + self.cursor.record_index = self.records.len(); + self.cursor.byte_offset = 0; + } + + pub(crate) fn read(&mut self, max_len: usize) -> Vec { + let mut out = Vec::new(); + while out.len() < max_len && self.cursor.record_index < self.records.len() { + match &self.records[self.cursor.record_index] { + TapeRecord::Filemark => { + self.cursor.record_index += 1; + self.cursor.byte_offset = 0; + break; + } + TapeRecord::Data(data) => { + let start = self.cursor.byte_offset.min(data.len()); + let remaining = max_len - out.len(); + let end = (start + remaining).min(data.len()); + out.extend_from_slice(&data[start..end]); + if end == data.len() { + self.cursor.record_index += 1; + self.cursor.byte_offset = 0; + } else { + self.cursor.byte_offset = end; + } + } + } + } + out + } + + pub(crate) fn seek_filemark(&mut self, mut count: u32) -> bool { + while count > 0 && self.cursor.record_index < self.records.len() { + match &self.records[self.cursor.record_index] { + TapeRecord::Filemark => { + count -= 1; + self.cursor.record_index += 1; + self.cursor.byte_offset = 0; + } + TapeRecord::Data(_) => { + self.cursor.record_index += 1; + self.cursor.byte_offset = 0; + } + } + } + count == 0 + } +} diff --git a/crates/vtl/src/simulator.rs b/crates/vtl/src/simulator.rs new file mode 100644 index 0000000..f9a2cc3 --- /dev/null +++ b/crates/vtl/src/simulator.rs @@ -0,0 +1,291 @@ +use crate::error::{Result, VtlError}; +use crate::interface::{MediumChanger, TapeDrive, TapeInventory}; +use crate::model::{ElementAddress, ElementKind, TapeBarcode, VirtualTape}; + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct Slot { + address: ElementAddress, + tape: Option, +} + +impl Slot { + fn new(address: ElementAddress) -> Self { + Self { + address, + tape: None, + } + } + + pub fn address(&self) -> ElementAddress { + self.address + } + + pub fn is_empty(&self) -> bool { + self.tape.is_none() + } + + pub fn barcode(&self) -> Option<&TapeBarcode> { + self.tape.as_ref().map(VirtualTape::barcode) + } +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct Drive { + address: ElementAddress, + tape: Option, +} + +impl Drive { + fn new(address: ElementAddress) -> Self { + Self { + address, + tape: None, + } + } + + pub fn address(&self) -> ElementAddress { + self.address + } + + pub fn is_empty(&self) -> bool { + self.tape.is_none() + } + + pub fn loaded_barcode(&self) -> Option<&TapeBarcode> { + self.tape.as_ref().map(VirtualTape::barcode) + } + + pub fn loaded_tape(&self) -> Option<&VirtualTape> { + self.tape.as_ref() + } +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct VirtualTapeLibrary { + slots: Vec, + drives: Vec, +} + +impl VirtualTapeLibrary { + pub fn new(slot_count: u32, drive_count: u32) -> Self { + let slots = (1..=slot_count) + .map(|index| Slot::new(ElementAddress::slot(index))) + .collect(); + let drives = (0..drive_count) + .map(|index| Drive::new(ElementAddress::drive(index))) + .collect(); + Self { slots, drives } + } + + pub fn slots(&self) -> &[Slot] { + &self.slots + } + + pub fn drives(&self) -> &[Drive] { + &self.drives + } + + pub fn slot(&self, address: ElementAddress) -> Result<&Slot> { + self.ensure_slot(address)?; + self.slots + .iter() + .find(|slot| slot.address == address) + .ok_or_else(|| VtlError::SlotOutOfRange(address.index())) + } + + pub fn drive(&self, address: ElementAddress) -> Result<&Drive> { + self.ensure_drive(address)?; + self.drives + .iter() + .find(|drive| drive.address == address) + .ok_or_else(|| VtlError::DriveOutOfRange(address.index())) + } + + fn slot_mut(&mut self, address: ElementAddress) -> Result<&mut Slot> { + self.ensure_slot(address)?; + self.slots + .iter_mut() + .find(|slot| slot.address == address) + .ok_or_else(|| VtlError::SlotOutOfRange(address.index())) + } + + fn drive_mut(&mut self, address: ElementAddress) -> Result<&mut Drive> { + self.ensure_drive(address)?; + self.drives + .iter_mut() + .find(|drive| drive.address == address) + .ok_or_else(|| VtlError::DriveOutOfRange(address.index())) + } + + fn ensure_slot(&self, address: ElementAddress) -> Result<()> { + if address.kind() != ElementKind::Slot { + return Err(VtlError::WrongElement { + expected: "slot", + actual: address.to_string(), + }); + } + if address.index() == 0 || address.index() as usize > self.slots.len() { + return Err(VtlError::SlotOutOfRange(address.index())); + } + Ok(()) + } + + fn ensure_drive(&self, address: ElementAddress) -> Result<()> { + if address.kind() != ElementKind::Drive { + return Err(VtlError::WrongElement { + expected: "drive", + actual: address.to_string(), + }); + } + if address.index() as usize >= self.drives.len() { + return Err(VtlError::DriveOutOfRange(address.index())); + } + Ok(()) + } + + fn take_from_slot(&mut self, slot: ElementAddress) -> Result { + self.slot_mut(slot)? + .tape + .take() + .ok_or_else(|| VtlError::SlotEmpty(slot.index())) + } + + fn put_into_slot(&mut self, slot: ElementAddress, tape: VirtualTape) -> Result<()> { + let slot_ref = self.slot_mut(slot)?; + if slot_ref.tape.is_some() { + return Err(VtlError::SlotOccupied(slot.index())); + } + slot_ref.tape = Some(tape); + Ok(()) + } + + fn take_from_drive(&mut self, drive: ElementAddress) -> Result { + self.drive_mut(drive)? + .tape + .take() + .ok_or_else(|| VtlError::DriveEmpty(drive.index())) + } + + fn put_into_drive(&mut self, drive: ElementAddress, tape: VirtualTape) -> Result<()> { + let drive_ref = self.drive_mut(drive)?; + if drive_ref.tape.is_some() { + return Err(VtlError::DriveOccupied(drive.index())); + } + drive_ref.tape = Some(tape); + Ok(()) + } + + fn loaded_tape_mut(&mut self, drive: ElementAddress) -> Result<&mut VirtualTape> { + self.drive_mut(drive)? + .tape + .as_mut() + .ok_or_else(|| VtlError::DriveEmpty(drive.index())) + } +} + +impl TapeInventory for VirtualTapeLibrary { + fn insert_tape(&mut self, slot: ElementAddress, barcode: TapeBarcode) -> Result<()> { + self.put_into_slot(slot, VirtualTape::new(barcode)) + } +} + +impl MediumChanger for VirtualTapeLibrary { + fn move_medium(&mut self, from: ElementAddress, to: ElementAddress) -> Result<()> { + match (from.kind(), to.kind()) { + (ElementKind::Slot, ElementKind::Slot) => { + let tape = self.take_from_slot(from)?; + self.put_into_slot(to, tape) + } + (ElementKind::Slot, ElementKind::Drive) => { + let tape = self.take_from_slot(from)?; + self.put_into_drive(to, tape) + } + (ElementKind::Drive, ElementKind::Slot) => { + let tape = self.take_from_drive(from)?; + self.put_into_slot(to, tape) + } + (ElementKind::Drive, ElementKind::Drive) => { + let tape = self.take_from_drive(from)?; + self.put_into_drive(to, tape) + } + _ => Err(VtlError::InvalidElementAddress(format!( + "unsupported move from {from} to {to}" + ))), + } + } + + fn load(&mut self, slot: ElementAddress, drive: ElementAddress) -> Result<()> { + self.move_medium(slot, drive) + } + + fn unload(&mut self, drive: ElementAddress, slot: ElementAddress) -> Result<()> { + self.move_medium(drive, slot) + } +} + +impl TapeDrive for VirtualTapeLibrary { + fn rewind(&mut self, drive: ElementAddress) -> Result<()> { + self.loaded_tape_mut(drive)?.rewind(); + Ok(()) + } + + fn seek_filemark(&mut self, drive: ElementAddress, count: u32) -> Result<()> { + if self.loaded_tape_mut(drive)?.seek_filemark(count) { + Ok(()) + } else { + Err(VtlError::FilemarkNotFound) + } + } + + fn write_filemark(&mut self, drive: ElementAddress) -> Result<()> { + self.loaded_tape_mut(drive)?.append_filemark(); + Ok(()) + } + + fn write(&mut self, drive: ElementAddress, data: &[u8]) -> Result<()> { + self.loaded_tape_mut(drive)?.append_data(data); + Ok(()) + } + + fn read(&mut self, drive: ElementAddress, max_len: usize) -> Result> { + Ok(self.loaded_tape_mut(drive)?.read(max_len)) + } +} + +impl VirtualTapeLibrary { + pub fn insert_tape(&mut self, slot: ElementAddress, barcode: TapeBarcode) -> Result<()> { + ::insert_tape(self, slot, barcode) + } + + pub fn move_medium(&mut self, from: ElementAddress, to: ElementAddress) -> Result<()> { + ::move_medium(self, from, to) + } + + pub fn load(&mut self, slot: ElementAddress, drive: ElementAddress) -> Result<()> { + ::load(self, slot, drive) + } + + pub fn unload(&mut self, drive: ElementAddress, slot: ElementAddress) -> Result<()> { + ::unload(self, drive, slot) + } + + pub fn rewind(&mut self, drive: ElementAddress) -> Result<()> { + ::rewind(self, drive) + } + + pub fn seek_filemark(&mut self, drive: ElementAddress, count: u32) -> Result<()> { + ::seek_filemark(self, drive, count) + } + + pub fn write_filemark(&mut self, drive: ElementAddress) -> Result<()> { + ::write_filemark(self, drive) + } + + pub fn write(&mut self, drive: ElementAddress, data: &[u8]) -> Result<()> { + ::write(self, drive, data) + } + + pub fn read(&mut self, drive: ElementAddress, max_len: usize) -> Result> { + ::read(self, drive, max_len) + } +} diff --git a/crates/vtl/tests/behavior.rs b/crates/vtl/tests/behavior.rs new file mode 100644 index 0000000..7ff2b73 --- /dev/null +++ b/crates/vtl/tests/behavior.rs @@ -0,0 +1,152 @@ +use coldstore_vtl::command::{CommandSpec, RecordedCommandRunner}; +use coldstore_vtl::discover::{parse_lsscsi, ScsiDeviceKind}; +use coldstore_vtl::mhvtl::MhvtlToolchain; +use coldstore_vtl::model::{ElementAddress, TapeBarcode}; +use coldstore_vtl::simulator::VirtualTapeLibrary; + +const LSSCSI_SAMPLE: &str = r#" +[5:0:0:0] mediumx STK L700 0106 /dev/sch0 /dev/sg8 +[5:0:1:0] tape IBM ULT3580-TD5 0106 /dev/st4 /dev/sg10 +[5:0:2:0] tape IBM ULT3580-TD5 0106 /dev/st1 /dev/sg6 +"#; + +#[test] +fn parses_lsscsi_tape_and_medium_changer_devices() { + let inventory = parse_lsscsi(LSSCSI_SAMPLE).expect("parse lsscsi output"); + + let changers = inventory.medium_changers(); + assert_eq!(changers.len(), 1); + assert_eq!(changers[0].kind, ScsiDeviceKind::MediumChanger); + assert_eq!(changers[0].hctl, "5:0:0:0"); + assert_eq!(changers[0].vendor, "STK"); + assert_eq!(changers[0].product, "L700"); + assert_eq!(changers[0].primary_device.as_deref(), Some("/dev/sch0")); + assert_eq!(changers[0].sg_device.as_deref(), Some("/dev/sg8")); + + let drives = inventory.tape_drives(); + assert_eq!(drives.len(), 2); + assert_eq!(drives[0].kind, ScsiDeviceKind::TapeDrive); + assert_eq!(drives[0].primary_device.as_deref(), Some("/dev/st4")); + assert_eq!(drives[0].non_rewinding_device.as_deref(), Some("/dev/nst4")); + assert_eq!(drives[0].sg_device.as_deref(), Some("/dev/sg10")); +} + +#[test] +fn mhvtl_toolchain_exposes_stable_lsscsi_mtx_mt_and_sg_command_paths() { + let tools = MhvtlToolchain::default(); + + assert_eq!( + tools.discover_command(), + CommandSpec::new("lsscsi").arg("-g") + ); + assert_eq!( + tools.library_status_command("/dev/sg8"), + CommandSpec::new("mtx").args(["-f", "/dev/sg8", "status"]) + ); + assert_eq!( + tools.move_medium_command( + "/dev/sg8", + ElementAddress::slot(1), + ElementAddress::drive(0) + ), + CommandSpec::new("mtx").args(["-f", "/dev/sg8", "transfer", "1", "0"]) + ); + assert_eq!( + tools.load_command( + "/dev/sg8", + ElementAddress::slot(2), + ElementAddress::drive(0) + ), + CommandSpec::new("mtx").args(["-f", "/dev/sg8", "load", "2", "0"]) + ); + assert_eq!( + tools.unload_command( + "/dev/sg8", + ElementAddress::slot(2), + ElementAddress::drive(0) + ), + CommandSpec::new("mtx").args(["-f", "/dev/sg8", "unload", "2", "0"]) + ); + assert_eq!( + tools.rewind_command("/dev/nst0"), + CommandSpec::new("mt").args(["-f", "/dev/nst0", "rewind"]) + ); + assert_eq!( + tools.write_filemark_command("/dev/nst0", 2), + CommandSpec::new("mt").args(["-f", "/dev/nst0", "weof", "2"]) + ); + assert_eq!( + tools.seek_filemark_command("/dev/nst0", 1), + CommandSpec::new("mt").args(["-f", "/dev/nst0", "fsf", "1"]) + ); + assert_eq!( + tools.test_unit_ready_command("/dev/sg10"), + CommandSpec::new("sg_turs").arg("/dev/sg10") + ); + assert_eq!( + tools.inquiry_command("/dev/sg10"), + CommandSpec::new("sg_inq").arg("/dev/sg10") + ); +} + +#[test] +fn mhvtl_discovery_uses_injected_runner_instead_of_host_commands() { + let tools = MhvtlToolchain::default(); + let runner = RecordedCommandRunner::default(); + runner.push_stdout(LSSCSI_SAMPLE); + + let inventory = tools + .discover(&runner) + .expect("discover from recorded output"); + + assert_eq!(inventory.tape_drives().len(), 2); + assert_eq!(runner.commands(), vec![tools.discover_command()]); +} + +#[test] +fn memory_vtl_models_changer_load_unload_and_tape_filemark_flow() { + let mut vtl = VirtualTapeLibrary::new(3, 1); + vtl.insert_tape(ElementAddress::slot(1), TapeBarcode::new("TAPE001")) + .expect("insert tape"); + + vtl.load(ElementAddress::slot(1), ElementAddress::drive(0)) + .expect("load tape"); + assert!(vtl.slot(ElementAddress::slot(1)).unwrap().is_empty()); + assert_eq!( + vtl.drive(ElementAddress::drive(0)) + .unwrap() + .loaded_barcode() + .map(|b| b.as_str()), + Some("TAPE001") + ); + + vtl.write(ElementAddress::drive(0), b"abc") + .expect("write abc"); + vtl.write_filemark(ElementAddress::drive(0)) + .expect("write filemark"); + vtl.write(ElementAddress::drive(0), b"def") + .expect("write def"); + + vtl.rewind(ElementAddress::drive(0)).expect("rewind"); + assert_eq!( + vtl.read(ElementAddress::drive(0), 3).expect("read abc"), + b"abc" + ); + vtl.seek_filemark(ElementAddress::drive(0), 1) + .expect("seek filemark"); + assert_eq!( + vtl.read(ElementAddress::drive(0), 3).expect("read def"), + b"def" + ); + + vtl.unload(ElementAddress::drive(0), ElementAddress::slot(2)) + .expect("unload tape"); + assert!(vtl.drive(ElementAddress::drive(0)).unwrap().is_empty()); + assert_eq!( + vtl.slot(ElementAddress::slot(2)) + .unwrap() + .barcode() + .map(|b| b.as_str()), + Some("TAPE001") + ); +} diff --git a/docs/modules/05-scheduler-layer.md b/docs/modules/05-scheduler-layer.md index f36d702..43044e0 100644 --- a/docs/modules/05-scheduler-layer.md +++ b/docs/modules/05-scheduler-layer.md @@ -7,6 +7,17 @@ 归档取回调度层负责冷数据的归档与取回调度，是 ColdStore 的核心业务逻辑层。设计需充分考虑**磁带的顺序读写特性**，通过聚合与合并策略最大化吞吐、减少换带与 seek。 +### 1.0 当前实现状态（Phase-1 安全基线） + +当前代码已经有一个可单测的安全归档批处理切片： + +- `MetadataBackedSchedulerBackend::archive_staging_batch(cache, tape, limit)` 通过注入的 `Phase1ArchiveCache` 和 `TapeArchiveWriter` 执行归档批处理，不直接访问宿主机设备。 +- `CacheArchiveClient` 已把真实 `CacheService` gRPC client 适配为 `Phase1ArchiveCache`；单测使用 in-process CacheService + 临时 HDD backend 验证 staging list/get/delete，而不是仅靠 fake cache。 +- `TapeArchiveClient` 已把真实 `TapeService` gRPC client 适配为 `TapeArchiveWriter`；单测使用 in-process TapeService + simulator backend 验证 `WriteBundle` filemark 写入与 `ReadBundle` 读回，而不是直接调用 service helper。 +- 该切片会读取 staging 数据、调用 tape writer 写入 bundle、写入 `ArchiveBundle`，再更新对象 `archive_id` / `tape_id` / `tape_set` / `tape_block_offset` 和 `StorageClass::Cold`，最后删除 staging 数据。 +- 单测使用 in-process `coldstore-tape` simulator writer / gRPC client 验证真实 filemark 写入与读回；没有启动 mhVTL、没有访问 `/dev/*`。 +- 当前一条 staging object 对应一个 Phase-1 bundle（`phase1-bundle:/`）；多对象聚合、驱动选择、失败补偿和 live tape worker client 仍是后续阶段。 + > **部署模型**：调度层运行在 **Scheduler Worker** 节点上，是 ColdStore 的**唯一业务中枢**。 > Gateway 的全部 S3 请求都发往 Scheduler Worker 处理。 > Scheduler 通过 gRPC 对接 Cache Worker（同机）和 Tape Worker（远程）， diff --git a/docs/modules/06-tape-layer.md b/docs/modules/06-tape-layer.md index 82440a3..e872e4d 100644 --- a/docs/modules/06-tape-layer.md +++ b/docs/modules/06-tape-layer.md @@ -7,6 +7,19 @@ 磁带管理层提供磁带设备与磁带库的完整管理能力，通过自研 SDK 抽象层实现与底层硬件的解耦，前期对接 Linux SCSI 协议。 +### 1.0 当前实现状态（Phase-1 安全基线） + +当前代码已经包含一个不触碰宿主机设备的 Phase-1 磁带闭环： + +- `crates/vtl` 提供独立的虚拟磁带库 harness，覆盖 `lsscsi -g` 输出解析、`mtx`/`mt`/`sg3_utils`/`dd` 命令规格构造、runner 注入，以及纯内存 `VirtualTapeLibrary`。 +- `crates/tape` 的 `TapeServiceImpl` 通过 `TapeBackend` trait 依赖后端抽象，默认使用 `SimulatorTapeBackend`，不会访问 `/dev/st*`、`/dev/nst*`、`/dev/sg*` 或 `/dev/sch*`。 +- 已通过 gRPC service 层暴露可单测的驱动/带库操作：`ListDrives`、`GetDriveStatus`、`AcquireDrive`、`ReleaseDrive`、`LoadTape`、`UnloadTape`、`Rewind`、`SeekToFilemark`、`GetTapeMediaStatus`、`Inventory`。 +- 已通过 service/helper 和真实 gRPC client 路径暴露 filemark 分隔的 bundle 写入与按 filemark 读取；Scheduler 侧的 `TapeArchiveClient` 可通过 `TapeServiceClient.WriteBundle` 写入 simulator-backed TapeService。 +- `ReadBundle` 的 block offset 定位仍返回显式 `UNIMPLEMENTED`，等待真实 SCSI/磁带定位语义接入。 +- mhVTL 安装脚本位于 `scripts/setup-mhvtl-env.sh`，默认 dry-run；当前安全验证只执行 `bash -n`，不安装系统包、不启动内核模块或服务。 + +真实 mhVTL/SCSI 闭环属于后续专用 VM/live 测试目标，必须显式使用 live runner 或安装脚本 `--execute`，不能作为默认单测路径。 + ### 1.1 完整职责 | 功能域 | 说明 | diff --git a/scripts/setup-mhvtl-env.sh b/scripts/setup-mhvtl-env.sh new file mode 100755 index 0000000..d31629e --- /dev/null +++ b/scripts/setup-mhvtl-env.sh @@ -0,0 +1,179 @@ +#!/usr/bin/env bash +# Prepare a dedicated Linux VM/test host for mhVTL live tape-library testing. +# +# SAFETY DEFAULT: dry-run only. Nothing is installed unless --execute is passed. +# Do NOT run this on a normal development host unless you intentionally want to +# install kernel modules, systemd services, and virtual SCSI tape/changer devices. + +set -euo pipefail + +EXECUTE=0 +START_SERVICES=0 +MHVTL_REPO="${MHVTL_REPO:-https://github.com/markh794/mhvtl.git}" +MHVTL_SRC="${MHVTL_SRC:-/usr/local/src/mhvtl}" + +usage() { + cat <<'USAGE' +Usage: + scripts/setup-mhvtl-env.sh [--execute] [--start-services] + +Default mode is dry-run: commands are printed but not executed. + +Options: + --execute Actually install packages, clone/build/install mhVTL. + --start-services After install, enable/start mhvtl.target via systemctl. + -h, --help Show this help. + +Environment: + MHVTL_REPO Git repo to clone. Default: https://github.com/markh794/mhvtl.git + MHVTL_SRC Source checkout path. Default: /usr/local/src/mhvtl + +Expected live validation after install: + lsscsi -g + ps -ax | grep '[v]tl' + mtx -f /dev/sgX status # use changer sg path from lsscsi output + mt -f /dev/nstX status # use non-rewinding tape path + sg_inq /dev/sgX + sg_turs /dev/sgX +USAGE +} + +while [[ $# -gt 0 ]]; do + case "$1" in + --execute) EXECUTE=1 ;; + --start-services) START_SERVICES=1 ;; + -h|--help) usage; exit 0 ;; + *) echo "unknown argument: $1" >&2; usage; exit 2 ;; + esac + shift +done + +if [[ $EXECUTE -eq 0 ]]; then + echo "DRY-RUN mode. Pass --execute to actually modify this host." >&2 +fi + +SUDO="" +if [[ ${EUID:-$(id -u)} -ne 0 ]]; then + SUDO="sudo" +fi + +run() { + printf '+ ' + printf '%q ' "$@" + printf '\n' + if [[ $EXECUTE -eq 1 ]]; then + "$@" + fi +} + +run_sh() { + printf '+ %s\n' "$*" + if [[ $EXECUTE -eq 1 ]]; then + bash -lc "$*" + fi +} + +require_linux() { + if [[ "$(uname -s)" != "Linux" ]]; then + echo "mhVTL requires Linux" >&2 + exit 1 + fi +} + +install_packages() { + if command -v apt-get >/dev/null 2>&1; then + run $SUDO apt-get update + run $SUDO apt-get install -y \ + build-essential git make gcc kmod pkg-config \ + "linux-headers-$(uname -r)" \ + lsscsi sg3-utils mtx mt-st \ + zlib1g-dev liblzo2-dev + elif command -v dnf >/dev/null 2>&1; then + run $SUDO dnf install -y \ + gcc gcc-c++ make git kmod kernel-devel kernel-headers \ + lsscsi sg3_utils mtx mt-st \ + zlib-devel lzo-devel + elif command -v yum >/dev/null 2>&1; then + run $SUDO yum install -y \ + gcc gcc-c++ make git kmod kernel-devel kernel-headers \ + lsscsi sg3_utils mtx mt-st \ + zlib-devel lzo-devel + elif command -v zypper >/dev/null 2>&1; then + run $SUDO zypper --non-interactive install \ + gcc gcc-c++ make git-core kmod kernel-default-devel \ + lsscsi sg3_utils mtx mt_st \ + zlib-devel lzo-devel + elif command -v pacman >/dev/null 2>&1; then + run $SUDO pacman -Sy --needed --noconfirm \ + base-devel git linux-headers kmod \ + lsscsi sg3_utils mtx mt-st \ + zlib lzo + else + echo "unsupported package manager; install build tools, kernel headers, lsscsi, sg3_utils, mtx, mt-st, zlib-dev, lzo-dev manually" >&2 + exit 1 + fi +} + +checkout_mhvtl() { + if [[ -d "$MHVTL_SRC/.git" ]]; then + run git -C "$MHVTL_SRC" fetch --all --tags --prune + else + run $SUDO mkdir -p "$(dirname "$MHVTL_SRC")" + if [[ -n "$SUDO" ]]; then + run $SUDO git clone "$MHVTL_REPO" "$MHVTL_SRC" + else + run git clone "$MHVTL_REPO" "$MHVTL_SRC" + fi + fi +} + +build_and_install_mhvtl() { + # Upstream layouts have changed over time. Build the kernel module directory + # explicitly when present, then run the top-level build/install. + if [[ -d "$MHVTL_SRC/kernel" ]]; then + run make -C "$MHVTL_SRC/kernel" + run $SUDO make -C "$MHVTL_SRC/kernel" install + fi + + run make -C "$MHVTL_SRC" + run $SUDO make -C "$MHVTL_SRC" install + run $SUDO systemctl daemon-reload +} + +start_services() { + if [[ $START_SERVICES -eq 1 ]]; then + run $SUDO systemctl enable mhvtl.target + run $SUDO systemctl start mhvtl.target + run $SUDO systemctl status --no-pager mhvtl.target + else + echo "Skipping service start. Pass --start-services to enable/start mhvtl.target." >&2 + fi +} + +print_next_steps() { + cat <<'NEXT' + +Next manual checks on the dedicated VM/test host: + lsscsi -g + ps -ax | grep '[v]tl' + +Pick paths from lsscsi output: + mediumx ... /dev/sch0 /dev/sgCHANGER + tape ... /dev/stN /dev/sgTAPE + +Then test command paths: + mtx -f /dev/sgCHANGER status + mt -f /dev/nstN status + sg_inq /dev/sgTAPE + sg_turs /dev/sgTAPE + +Do not point ColdStore live tests at production tape devices. +NEXT +} + +require_linux +install_packages +checkout_mhvtl +build_and_install_mhvtl +start_services +print_next_steps