diff --git a/seaweed-volume/src/remote_storage/s3_tier.rs b/seaweed-volume/src/remote_storage/s3_tier.rs index 1bcea50bb7..98603fb25f 100644 --- a/seaweed-volume/src/remote_storage/s3_tier.rs +++ b/seaweed-volume/src/remote_storage/s3_tier.rs @@ -5,21 +5,153 @@ use std::collections::HashMap; use std::future::Future; +use std::pin::Pin; use std::sync::{Arc, OnceLock, RwLock}; +use std::task::{Context, Poll}; +use std::time::Duration; use aws_sdk_s3::Client; use aws_sdk_s3::config::http::HttpResponse; +use aws_sdk_s3::config::timeout::TimeoutConfig; use aws_sdk_s3::config::{BehaviorVersion, Credentials, Region}; use aws_sdk_s3::error::{DisplayErrorContext, SdkError}; use aws_sdk_s3::operation::get_object::GetObjectError; use aws_sdk_s3::operation::head_object::HeadObjectError; +use aws_sdk_s3::operation::upload_part::UploadPartOutput; +use aws_sdk_s3::operation::upload_part::builders::UploadPartFluentBuilder; +use aws_sdk_s3::primitives::{ByteStream, SdkBody}; use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart}; +use bytes::Bytes; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; -use tokio::sync::Semaphore; +use tokio::sync::{Semaphore, watch}; /// Concurrency limit for multipart upload/download (matches Go's s3manager). const CONCURRENCY: usize = 5; +/// A tier endpoint that drops packets or accepts connections without answering +/// must not hold a read, or a volume server's startup, for minutes per request. +/// Matches Go's s3ConnectTimeout and s3ReadTimeout. The SDK's stalled-stream +/// protection already fails a body that stops arriving. +const S3_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +const S3_READ_TIMEOUT: Duration = Duration::from_secs(60); + +/// The SDK's read timeout runs from the start of the request, so on +/// upload_part it would also cap sending the part body. send_part bounds the +/// response wait instead, from when the body is written, like Go's +/// ResponseHeaderTimeout. +fn upload_part_config() -> aws_sdk_s3::config::Builder { + aws_sdk_s3::config::Builder::default().timeout_config( + TimeoutConfig::builder() + .connect_timeout(S3_CONNECT_TIMEOUT) + .disable_read_timeout() + .build(), + ) +} + +/// A part goes to the connection in chunks this size, so its last chunk is +/// taken only once nearly all of the part has been written. +const UPLOAD_CHUNK_SIZE: usize = 64 * 1024; + +/// The latest attempt to send a part, and whether its body has been sent. +type PartSent = Arc>; + +struct PartBody { + rest: Bytes, + attempt: u64, + sent: PartSent, +} + +impl http_body::Body for PartBody { + type Data = Bytes; + type Error = std::convert::Infallible; + + fn poll_frame( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + if self.rest.is_empty() { + return Poll::Ready(None); + } + let n = self.rest.len().min(UPLOAD_CHUNK_SIZE); + let chunk = self.rest.split_to(n); + if self.rest.is_empty() { + let attempt = self.attempt; + self.sent.send_if_modified(|state| { + let latest = state.0 == attempt; + state.1 |= latest; + latest + }); + } + Poll::Ready(Some(Ok(http_body::Frame::data(chunk)))) + } + + fn is_end_stream(&self) -> bool { + self.rest.is_empty() + } + + fn size_hint(&self) -> http_body::SizeHint { + http_body::SizeHint::with_exact(self.rest.len() as u64) + } +} + +/// Sends one multipart part, giving up once `response_timeout` passes after +/// the whole part was handed to the connection without a response. A retry +/// rebuilds the body and restarts that clock. The body keeps its in-memory +/// contents, so signing and the checksum header are unchanged. +async fn send_part( + request: UploadPartFluentBuilder, + data: Bytes, + response_timeout: Duration, +) -> Result { + let sent: PartSent = Arc::new(watch::channel((0, false)).0); + let body = { + let sent = sent.clone(); + SdkBody::from(data.clone()).map_preserve_contents(move |_| { + let mut attempt = 0; + sent.send_modify(|state| { + *state = (state.0 + 1, data.is_empty()); + attempt = state.0; + }); + SdkBody::from_body_1_x(PartBody { + rest: data.clone(), + attempt, + sent: sent.clone(), + }) + }) + }; + tokio::select! { + resp = request + .body(ByteStream::new(body)) + .customize() + .config_override(upload_part_config()) + .send() => resp.map_err(|e| DisplayErrorContext(&e).to_string()), + () = wait_unanswered(sent, response_timeout) => Err(format!( + "no response within {response_timeout:?} after sending the part" + )), + } +} + +/// Completes once an attempt's body has been sent and `response_timeout` +/// passes before a later attempt starts. +async fn wait_unanswered(sent: PartSent, response_timeout: Duration) { + let mut state = sent.subscribe(); + loop { + let attempt = state + .wait_for(|&(_, done)| done) + .await + .map_or(0, |state| state.0); + if tokio::time::timeout( + response_timeout, + state.wait_for(|&(latest, _)| latest != attempt), + ) + .await + .is_err() + { + return; + } + } +} + /// A tier transfer failure. The variant is what callers match on; the /// message is the operator-facing text. #[derive(Debug, thiserror::Error)] @@ -107,7 +239,13 @@ impl S3TierBackend { .behavior_version(BehaviorVersion::latest()) .region(Region::new(region.to_string())) .credentials_provider(credentials) - .force_path_style(config.force_path_style); + .force_path_style(config.force_path_style) + .timeout_config( + TimeoutConfig::builder() + .connect_timeout(S3_CONNECT_TIMEOUT) + .read_timeout(S3_READ_TIMEOUT) + .build(), + ); if !config.endpoint.is_empty() { s3_config = s3_config.endpoint_url(&config.endpoint); @@ -231,23 +369,23 @@ impl S3TierBackend { TierError::Io(format!("failed to read file at offset {}: {}", off, e)) })?; - let upload_part_resp = client - .upload_part() - .bucket(&bucket) - .key(&key) - .upload_id(&upload_id) - .part_number(pn) - .body(buf.into()) - .send() - .await - .map_err(|e| { - TierError::Io(format!( - "failed to upload part {} at offset {}: {}", - pn, - off, - DisplayErrorContext(&e) - )) - })?; + let upload_part_resp = send_part( + client + .upload_part() + .bucket(&bucket) + .key(&key) + .upload_id(&upload_id) + .part_number(pn), + buf.into(), + S3_READ_TIMEOUT, + ) + .await + .map_err(|e| { + TierError::Io(format!( + "failed to upload part {} at offset {}: {}", + pn, off, e + )) + })?; let e_tag = upload_part_resp.e_tag().unwrap_or_default().to_string(); @@ -560,6 +698,25 @@ impl S3TierBackend { }) } + /// `read_range_blocking` that gives up once `timeout` passes, returning + /// `Ok(None)`. + pub fn read_range_blocking_within( + self: &Arc, + key: &str, + offset: u64, + size: usize, + timeout: Duration, + ) -> Result>, TierError> { + let backend = Arc::clone(self); + let key = key.to_string(); + block_on_tier_future(async move { + match tokio::time::timeout(timeout, backend.read_range(&key, offset, size)).await { + Ok(read) => read.map(Some), + Err(_) => Ok(None), + } + }) + } + pub fn read_range_blocking( &self, key: &str, @@ -725,6 +882,13 @@ fn describe_join_error(join_error: tokio::task::JoinError) -> TierError { mod tests { use super::*; use crate::remote_storage::s3::tests::{CannedResponse, NO_SUCH_KEY}; + use aws_sdk_s3::config::http::HttpRequest; + use aws_sdk_s3::config::{HttpClient, RuntimeComponents}; + use aws_smithy_runtime_api::client::http::{ + HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpConnector, + }; + use aws_smithy_runtime_api::http::StatusCode; + use http_body::Body as _; use std::collections::HashSet; use tokio::runtime::Handle; @@ -897,6 +1061,235 @@ mod tests { assert!(matches!(err, TierError::Io(_)), "{err:?}"); } + #[test] + fn backend_bounds_connect_and_read_waits() { + let (_listener, backend) = silent_backend(); + let timeouts = backend.client.config().timeout_config().unwrap(); + assert_eq!(timeouts.connect_timeout(), Some(S3_CONNECT_TIMEOUT)); + assert_eq!(timeouts.read_timeout(), Some(S3_READ_TIMEOUT)); + } + + #[derive(Debug)] + struct CaptureTimeouts(Arc>>); + + impl aws_sdk_s3::config::Intercept for CaptureTimeouts { + fn name(&self) -> &'static str { + "CaptureTimeouts" + } + + fn read_before_transmit( + &self, + _context: &aws_sdk_s3::config::interceptors::BeforeTransmitInterceptorContextRef<'_>, + _runtime_components: &aws_sdk_s3::config::RuntimeComponents, + cfg: &mut aws_sdk_s3::config::ConfigBag, + ) -> Result<(), aws_sdk_s3::error::BoxError> { + *self.0.lock().unwrap() = cfg.load::().cloned(); + Err("captured".into()) + } + } + + #[tokio::test] + async fn upload_part_keeps_the_connect_timeout_without_the_read_timeout() { + let (_listener, backend) = silent_backend(); + let seen = Arc::new(std::sync::Mutex::new(None)); + let _ = backend + .client + .upload_part() + .bucket("bucket") + .key("key") + .upload_id("upload") + .part_number(1) + .body(vec![0u8; 8].into()) + .customize() + .config_override(upload_part_config().interceptor(CaptureTimeouts(seen.clone()))) + .send() + .await; + let timeouts = seen + .lock() + .unwrap() + .clone() + .expect("request never reached transmit"); + assert_eq!(timeouts.connect_timeout(), Some(S3_CONNECT_TIMEOUT)); + assert_eq!(timeouts.read_timeout(), None); + } + + fn upload_part_request(backend: &S3TierBackend) -> UploadPartFluentBuilder { + backend + .client + .upload_part() + .bucket("bucket") + .key("key") + .upload_id("upload") + .part_number(1) + } + + #[tokio::test] + async fn send_part_gives_up_when_no_response_follows_the_body() { + let (_listener, backend) = silent_backend(); + let sent = tokio::time::timeout( + Duration::from_secs(30), + send_part( + upload_part_request(&backend), + Bytes::from_static(&[0u8; 8]), + Duration::from_millis(200), + ), + ) + .await + .expect("a part with no response did not give up"); + let err = sent.unwrap_err(); + assert!(err.starts_with("no response within 200ms"), "{err}"); + } + + #[tokio::test(start_paused = true)] + async fn a_retry_restarts_the_response_clock() { + let sent: PartSent = Arc::new(watch::channel((1, true)).0); + let mut unanswered = + std::pin::pin!(wait_unanswered(sent.clone(), Duration::from_millis(200))); + let wait = |ms| tokio::time::sleep(Duration::from_millis(ms)); + tokio::select! { + () = unanswered.as_mut() => panic!("gave up early"), + () = wait(150) => {} + } + // The retry starts and finishes sending before the waiter runs again. + sent.send_replace((2, false)); + sent.send_replace((2, true)); + tokio::select! { + () = unanswered.as_mut() => panic!("kept the first attempt's clock"), + () = wait(150) => {} + } + tokio::select! { + () = unanswered => {} + () = wait(100) => panic!("the retry never timed out"), + } + } + + /// Takes the request body one chunk per `pause`, then answers at once. + #[derive(Debug, Clone)] + struct SlowReader { + pause: Duration, + } + + impl HttpConnector for SlowReader { + fn call(&self, request: HttpRequest) -> HttpConnectorFuture { + let pause = self.pause; + HttpConnectorFuture::new(async move { + let mut body = std::pin::pin!(request.into_body()); + loop { + tokio::time::sleep(pause).await; + let frame = std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await; + if frame.is_none() || body.is_end_stream() { + break; + } + } + let mut response = + HttpResponse::new(StatusCode::try_from(200).unwrap(), SdkBody::empty()); + response.headers_mut().insert("ETag", "\"etag\""); + Ok(response) + }) + } + } + + impl HttpClient for SlowReader { + fn http_connector( + &self, + _settings: &HttpConnectorSettings, + _components: &RuntimeComponents, + ) -> SharedHttpConnector { + SharedHttpConnector::new(self.clone()) + } + } + + #[tokio::test] + async fn send_part_does_not_cap_a_slow_body() { + let config = aws_sdk_s3::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .region(Region::new("us-east-1")) + .credentials_provider(Credentials::new("AKIATEST", "secret", None, None, "test")) + .endpoint_url("http://127.0.0.1:1") + .force_path_style(true) + .http_client(SlowReader { + pause: Duration::from_millis(300), + }) + .build(); + let backend = S3TierBackend { + client: Client::from_conf(config), + bucket: "bucket".to_string(), + storage_class: "STANDARD".to_string(), + }; + let output = send_part( + upload_part_request(&backend), + Bytes::from(vec![0u8; 4 * UPLOAD_CHUNK_SIZE]), + Duration::from_millis(200), + ) + .await + .expect("a part still being sent was cut short"); + assert_eq!(output.e_tag(), Some("\"etag\"")); + } + + #[test] + fn read_range_blocking_within_gives_up_on_a_silent_endpoint() { + let (_listener, backend) = silent_backend(); + let start = std::time::Instant::now(); + let got = backend + .read_range_blocking_within("key", 0, 8, Duration::from_millis(200)) + .unwrap(); + assert!(got.is_none()); + assert!(start.elapsed() < Duration::from_secs(5)); + } + + #[tokio::test] + async fn read_range_gives_up_when_the_body_stalls() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + std::thread::spawn(move || { + let mut held = Vec::new(); + for mut conn in listener.incoming().flatten() { + let mut request = Vec::new(); + let mut byte = [0u8; 1]; + while !request.ends_with(b"\r\n\r\n") { + match std::io::Read::read(&mut conn, &mut byte) { + Ok(1) => request.push(byte[0]), + _ => break, + } + } + let _ = std::io::Write::write_all( + &mut conn, + b"HTTP/1.1 206 Partial Content\r\nContent-Length: 8\r\nContent-Range: bytes 0-7/1024\r\n\r\n", + ); + held.push(conn); + } + }); + let backend = S3TierBackend::new(&S3TierConfig { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + region: "us-east-1".to_string(), + bucket: "bucket".to_string(), + endpoint, + storage_class: String::new(), + force_path_style: true, + }); + + let read = tokio::time::timeout(Duration::from_secs(30), backend.read_range("key", 0, 8)) + .await + .expect("a read whose body stalls did not give up"); + let err = read.unwrap_err(); + assert!(matches!(err, TierError::Io(_)), "{err:?}"); + } + + fn silent_backend() -> (std::net::TcpListener, Arc) { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let backend = S3TierBackend::new(&S3TierConfig { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + region: "us-east-1".to_string(), + bucket: "bucket".to_string(), + endpoint: format!("http://{}", listener.local_addr().unwrap()), + storage_class: String::new(), + force_path_style: true, + }); + (listener, Arc::new(backend)) + } + #[test] fn read_range_blocking_no_such_key_is_not_found() { let err = backend_answering(404, NO_SUCH_KEY) diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 8fc3b7214a..b3824380b8 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -9,14 +9,14 @@ //! Matches Go's storage/volume.go, volume_loading.go, volume_read.go, //! volume_write.go, volume_super_block.go. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::fs::{self, File, OpenOptions}; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::ops::ControlFlow; use std::path::Path; use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicU64, Ordering}; -use std::sync::{Arc, Condvar, Mutex, RwLock}; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::sync::{Arc, Condvar, LazyLock, Mutex, RwLock}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tracing::{debug, error, info, warn}; @@ -1198,6 +1198,33 @@ impl RemoteDatFile { .backend .read_range_blocking(&self.key, offset, buf.len()) .map_err(io::Error::other)?; + Self::fill(buf, offset, data) + } + + /// `read_exact_at` that fails with `TimedOut` once `timeout` passes. + pub(crate) fn read_exact_at_within( + &self, + buf: &mut [u8], + offset: u64, + timeout: Duration, + ) -> io::Result<()> { + match self + .backend + .read_range_blocking_within(&self.key, offset, buf.len(), timeout) + .map_err(io::Error::other)? + { + Some(data) => Self::fill(buf, offset, data), + None => Err(io::Error::new( + io::ErrorKind::TimedOut, + format!( + "remote read at offset {} got no answer within {:?}", + offset, timeout + ), + )), + } + } + + fn fill(buf: &mut [u8], offset: u64, data: Vec) -> io::Result<()> { if data.len() != buf.len() { return Err(io::Error::new( io::ErrorKind::UnexpectedEof, @@ -1214,6 +1241,71 @@ impl RemoteDatFile { } } +/// A tiered volume loads without its superblock when the backend cannot serve +/// it; once a read stalls, the backend's other volumes skip it for a while. +/// Matches Go's remoteSuperBlockTimeout and remoteSuperBlockStallPeriod. +const REMOTE_SUPER_BLOCK_TIMEOUT: Duration = Duration::from_secs(10); +const REMOTE_SUPER_BLOCK_STALL_PERIOD: Duration = Duration::from_secs(60); +static REMOTE_SUPER_BLOCK_STALLS: LazyLock>> = + LazyLock::new(Default::default); + +fn read_super_block_bytes( + mut read: impl FnMut(&mut [u8], u64) -> Result<(), VolumeError>, +) -> Result, VolumeError> { + let mut header = [0u8; SUPER_BLOCK_SIZE]; + read(&mut header, 0)?; + + let extra_size = u16::from_be_bytes([header[6], header[7]]); + let mut full_buf = vec![0u8; SUPER_BLOCK_SIZE + extra_size as usize]; + full_buf[..SUPER_BLOCK_SIZE].copy_from_slice(&header); + if extra_size > 0 { + read(&mut full_buf[SUPER_BLOCK_SIZE..], SUPER_BLOCK_SIZE as u64)?; + } + Ok(full_buf) +} + +fn read_remote_super_block( + storage_name: &str, + remote: &RemoteDatFile, + timeout: Duration, + stall_period: Duration, +) -> Result, VolumeError> { + if let Some(until) = REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().get(storage_name) + && Instant::now() < *until + { + return Err(VolumeError::Io(io::Error::other(format!( + "backend {} stalled on a super block read, skipped", + storage_name + )))); + } + let deadline = Instant::now() + timeout; + let result = read_super_block_bytes(|buf, offset| { + remote + .read_exact_at_within( + buf, + offset, + deadline.saturating_duration_since(Instant::now()), + ) + .map_err(VolumeError::from) + }); + match &result { + Ok(_) => { + REMOTE_SUPER_BLOCK_STALLS + .lock() + .unwrap() + .remove(storage_name); + } + Err(VolumeError::Io(e)) if e.kind() == io::ErrorKind::TimedOut => { + REMOTE_SUPER_BLOCK_STALLS + .lock() + .unwrap() + .insert(storage_name.to_string(), Instant::now() + stall_period); + } + Err(_) => {} + } + result +} + // ============================================================================ // Volume // ============================================================================ @@ -1684,6 +1776,7 @@ impl Volume { ))); } + let mut remote_super_block_failed = false; if already_has_super_block { match self.read_super_block(io) { Ok(()) => { @@ -1700,6 +1793,12 @@ impl Volume { error = %e, "failed to read remote super block during load" ); + remote_super_block_failed = true; + let replication = self.volume_info.read().unwrap().replication.clone(); + if !replication.is_empty() { + self.super_block.write().unwrap().replica_placement = + ReplicaPlacement::from_string(&replication)?; + } } Err(e) => return Err(e), } @@ -1748,17 +1847,24 @@ impl Volume { // at the front of .idx. Best effort: a volume that cannot be // repaired is still servable for everything the surviving rows // index. - match self.repair_idx_head_tombstones(io) { - Ok(0) => {} - Ok(restored) => info!( + if remote_super_block_failed { + debug!( volume_id = self.id.0, - restored, "recovered overwritten .idx rows from .dat" - ), - Err(e) => warn!( - volume_id = self.id.0, - error = %e, - "recover overwritten .idx rows" - ), + "remote backend unavailable, not recovering overwritten .idx rows on this load" + ); + } else { + match self.repair_idx_head_tombstones(io) { + Ok(0) => {} + Ok(restored) => info!( + volume_id = self.id.0, + restored, "recovered overwritten .idx rows from .dat" + ), + Err(e) => warn!( + volume_id = self.id.0, + error = %e, + "recover overwritten .idx rows" + ), + } } // The permission bits still allow writing on a read-only mount; @@ -2220,21 +2326,15 @@ impl Volume { // ---- SuperBlock I/O ---- fn read_super_block(&self, io: &VolumeIo) -> Result<(), VolumeError> { - let mut header = [0u8; SUPER_BLOCK_SIZE]; - self.read_exact_at_backend(io, &mut header, 0)?; - - let extra_size = u16::from_be_bytes([header[6], header[7]]); - let total_size = SUPER_BLOCK_SIZE + extra_size as usize; - - let mut full_buf = vec![0u8; total_size]; - full_buf[..SUPER_BLOCK_SIZE].copy_from_slice(&header); - if extra_size > 0 { - self.read_exact_at_backend( - io, - &mut full_buf[SUPER_BLOCK_SIZE..], - SUPER_BLOCK_SIZE as u64, - )?; - } + let full_buf = match io.remote_dat_file.as_ref() { + Some(remote) if self.has_remote_file() => read_remote_super_block( + &self.remote_storage_name_key().0, + remote, + REMOTE_SUPER_BLOCK_TIMEOUT, + REMOTE_SUPER_BLOCK_STALL_PERIOD, + )?, + _ => read_super_block_bytes(|buf, offset| self.read_exact_at_backend(io, buf, offset))?, + }; *self.super_block.write().unwrap() = SuperBlock::from_bytes(&full_buf)?; @@ -11516,6 +11616,254 @@ mod tests { assert!(result.is_err(), "unopenable .dat must fail load"); } + fn silent_tier_backend() -> ( + std::net::TcpListener, + Arc, + ) { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let backend = crate::remote_storage::s3_tier::S3TierBackend::new( + &crate::remote_storage::s3_tier::S3TierConfig { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + region: "us-east-1".to_string(), + bucket: "bucket-a".to_string(), + endpoint: format!("http://{}", listener.local_addr().unwrap()), + storage_class: "STANDARD".to_string(), + force_path_style: true, + }, + ); + (listener, Arc::new(backend)) + } + + #[test] + fn remote_super_block_read_gives_up_on_a_silent_backend_and_skips_it() { + let name = "s3.super_block_stall_test"; + let (_listener, backend) = silent_tier_backend(); + let silent = RemoteDatFile { + backend, + key: "remote-key".to_string(), + file_size: 1024, + modified_time: 0, + }; + + let start = Instant::now(); + let err = read_remote_super_block( + name, + &silent, + Duration::from_millis(200), + Duration::from_secs(3600), + ) + .unwrap_err(); + assert!( + matches!(&err, VolumeError::Io(e) if e.kind() == io::ErrorKind::TimedOut), + "{err:?}" + ); + assert!(start.elapsed() < Duration::from_secs(5)); + + let start = Instant::now(); + assert!( + read_remote_super_block( + name, + &silent, + Duration::from_secs(30), + Duration::from_secs(3600) + ) + .is_err() + ); + assert!( + start.elapsed() < Duration::from_secs(1), + "a stalled backend must not be asked again within the stall period" + ); + + let sb = SuperBlock { + version: Version::current(), + ..SuperBlock::default() + }; + let (endpoint, shutdown_tx) = spawn_fake_s3_server(sb.to_bytes()); + let answering = RemoteDatFile { + backend: Arc::new(crate::remote_storage::s3_tier::S3TierBackend::new( + &crate::remote_storage::s3_tier::S3TierConfig { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + region: "us-east-1".to_string(), + bucket: "bucket-a".to_string(), + endpoint, + storage_class: "STANDARD".to_string(), + force_path_style: true, + }, + )), + key: "remote-key".to_string(), + file_size: SUPER_BLOCK_SIZE as u64, + modified_time: 0, + }; + REMOTE_SUPER_BLOCK_STALLS + .lock() + .unwrap() + .insert(name.to_string(), Instant::now()); + let bytes = read_remote_super_block( + name, + &answering, + Duration::from_secs(10), + Duration::from_secs(3600), + ) + .unwrap(); + assert_eq!( + SuperBlock::from_bytes(&bytes).unwrap().version, + Version::current() + ); + assert!( + !REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().contains_key(name), + "a successful read clears the stall" + ); + let _ = shutdown_tx.send(()); + } + + #[test] + fn remote_volume_loads_without_waiting_on_a_stalled_backend() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let dat_len = { + let v = make_test_volume(dir); + v.dat_file_size().unwrap() + }; + std::fs::remove_file(format!("{}/1.dat", dir)).unwrap(); + + let (_listener, backend) = silent_tier_backend(); + let name = "s3.load_stall_test"; + crate::remote_storage::s3_tier::global_s3_tier_registry() + .write() + .unwrap() + .register(name.to_string(), Arc::try_unwrap(backend).ok().unwrap()); + REMOTE_SUPER_BLOCK_STALLS + .lock() + .unwrap() + .insert(name.to_string(), Instant::now() + Duration::from_secs(3600)); + + let vif = VifVolumeInfo { + files: vec![VifRemoteFile { + backend_type: "s3".to_string(), + backend_id: "load_stall_test".to_string(), + key: "remote-key".to_string(), + offset: 0, + file_size: dat_len, + modified_time: 123, + extension: ".dat".to_string(), + }], + version: Version::current().0 as u32, + replication: "001".to_string(), + ..VifVolumeInfo::default() + }; + std::fs::write( + format!("{}/1.vif", dir), + serde_json::to_string_pretty(&vif).unwrap(), + ) + .unwrap(); + + let start = Instant::now(); + let v = Volume::new( + dir, + dir, + VolumeId(1), + NeedleMapKind::InMemory, + &VolumeSpec::default(), + ) + .unwrap(); + assert!(start.elapsed() < Duration::from_secs(5)); + assert!(v.has_remote_file()); + assert_eq!(v.version(), Version::current()); + assert_eq!( + v.super_block.read().unwrap().replica_placement, + ReplicaPlacement::from_string("001").unwrap() + ); + + crate::remote_storage::s3_tier::global_s3_tier_registry() + .write() + .unwrap() + .remove(name); + REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().remove(name); + } + + #[test] + fn remote_volume_load_skips_idx_repair_on_a_stalled_backend() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let dat_len = { + let v = make_test_volume(dir); + for i in 1..=3u64 { + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(0x1234), + data: format!("needle-{i}").into_bytes(), + data_size: format!("needle-{i}").len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + } + v.dat_file_size().unwrap() + }; + std::fs::remove_file(format!("{}/1.dat", dir)).unwrap(); + let idx_path = format!("{}/1.idx", dir); + crate::storage::volume_idx_repair::tests::clobber_idx_head(&idx_path, &[4, 5]); + let idx_before = std::fs::read(&idx_path).unwrap(); + + let (_listener, backend) = silent_tier_backend(); + let name = "s3.idx_repair_stall_test"; + crate::remote_storage::s3_tier::global_s3_tier_registry() + .write() + .unwrap() + .register(name.to_string(), Arc::try_unwrap(backend).ok().unwrap()); + REMOTE_SUPER_BLOCK_STALLS + .lock() + .unwrap() + .insert(name.to_string(), Instant::now() + Duration::from_secs(3600)); + + let vif = VifVolumeInfo { + files: vec![VifRemoteFile { + backend_type: "s3".to_string(), + backend_id: "idx_repair_stall_test".to_string(), + key: "remote-key".to_string(), + offset: 0, + file_size: dat_len, + modified_time: 123, + extension: ".dat".to_string(), + }], + version: Version::current().0 as u32, + ..VifVolumeInfo::default() + }; + std::fs::write( + format!("{}/1.vif", dir), + serde_json::to_string_pretty(&vif).unwrap(), + ) + .unwrap(); + + let start = Instant::now(); + let v = Volume::new( + dir, + dir, + VolumeId(1), + NeedleMapKind::InMemory, + &VolumeSpec::default(), + ) + .unwrap(); + assert!( + start.elapsed() < Duration::from_secs(5), + "the load waited on the stalled backend" + ); + assert!(v.has_remote_file()); + drop(v); + assert_eq!( + std::fs::read(&idx_path).unwrap(), + idx_before, + "the repair ran against an unreachable backend" + ); + + crate::remote_storage::s3_tier::global_s3_tier_registry() + .write() + .unwrap() + .remove(name); + REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().remove(name); + } + #[test] fn test_remote_only_volume_load_reads_from_tier_backend() { let tmp = TempDir::new().unwrap(); diff --git a/seaweed-volume/src/storage/volume_idx_repair.rs b/seaweed-volume/src/storage/volume_idx_repair.rs index e79e222633..0fd410db8a 100644 --- a/seaweed-volume/src/storage/volume_idx_repair.rs +++ b/seaweed-volume/src/storage/volume_idx_repair.rs @@ -224,7 +224,7 @@ fn finish_walk(walk: io::Result<()>) -> Result<(), VolumeError> { } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; use crate::storage::needle::crc::CRC; use crate::storage::needle_map::NeedleMapKind; @@ -267,7 +267,7 @@ mod tests { /// Reproduce the damage a delete on a tiered read-only volume used to do: it /// writes (key, offset 0, tombstone) rows over the front of .idx instead of /// appending them. - fn clobber_idx_head(idx_path: &str, keys: &[u64]) { + pub(crate) fn clobber_idx_head(idx_path: &str, keys: &[u64]) { let mut file = OpenOptions::new().write(true).open(idx_path).unwrap(); for (i, key) in keys.iter().enumerate() { let mut row = Vec::new(); diff --git a/weed/storage/backend/s3_backend/s3_backend.go b/weed/storage/backend/s3_backend/s3_backend.go index 08651dedea..689286ad0c 100644 --- a/weed/storage/backend/s3_backend/s3_backend.go +++ b/weed/storage/backend/s3_backend/s3_backend.go @@ -1,6 +1,7 @@ package s3_backend import ( + "context" "fmt" "io" "math" @@ -222,7 +223,9 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n bytesRange := fmt.Sprintf("bytes=%d-%d", off, off+int64(len(p))-1) - getObjectOutput, getObjectErr := s3backendStorageFile.backendStorage.conn.GetObject(&s3.GetObjectInput{ + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + getObjectOutput, getObjectErr := s3backendStorageFile.backendStorage.conn.GetObjectWithContext(ctx, &s3.GetObjectInput{ Bucket: &s3backendStorageFile.backendStorage.bucket, Key: &s3backendStorageFile.key, Range: &bytesRange, @@ -232,6 +235,8 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n return 0, fmt.Errorf("bucket %s GetObject %s: %v", s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, getObjectErr) } defer getObjectOutput.Body.Close() + stalled := time.AfterFunc(s3ReadTimeout, cancel) + defer stalled.Stop() // glog.V(3).Infof("read %s %s", s3backendStorageFile.key, bytesRange) // glog.V(3).Infof("content range: %s, contentLength: %d", *getObjectOutput.ContentRange, *getObjectOutput.ContentLength) @@ -242,6 +247,7 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n if len(p) == 0 { break } + stalled.Reset(s3ReadTimeout) readCount, err = getObjectOutput.Body.Read(p) n += readCount @@ -253,6 +259,9 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n if err == io.EOF && n == requested { err = nil } + if err != nil && ctx.Err() != nil { + err = fmt.Errorf("bucket %s read %s: no data for %v: %w", s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, s3ReadTimeout, err) + } return } diff --git a/weed/storage/backend/s3_backend/s3_backend_test.go b/weed/storage/backend/s3_backend/s3_backend_test.go index 347c379820..f103469ad0 100644 --- a/weed/storage/backend/s3_backend/s3_backend_test.go +++ b/weed/storage/backend/s3_backend/s3_backend_test.go @@ -7,6 +7,8 @@ import ( "testing" "testing/iotest" + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3iface" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" @@ -23,7 +25,7 @@ type stubS3Client struct { getObject func(*s3.GetObjectInput) (*s3.GetObjectOutput, error) } -func (s *stubS3Client) GetObject(input *s3.GetObjectInput) (*s3.GetObjectOutput, error) { +func (s *stubS3Client) GetObjectWithContext(_ aws.Context, input *s3.GetObjectInput, _ ...request.Option) (*s3.GetObjectOutput, error) { return s.getObject(input) } diff --git a/weed/storage/backend/s3_backend/s3_sessions.go b/weed/storage/backend/s3_backend/s3_sessions.go index bdadb56118..bf9f3af5f0 100644 --- a/weed/storage/backend/s3_backend/s3_sessions.go +++ b/weed/storage/backend/s3_backend/s3_sessions.go @@ -3,6 +3,7 @@ package s3_backend import ( "crypto/tls" "fmt" + "net" "net/http" "sync" "time" @@ -21,8 +22,14 @@ var ( sessionsLock sync.Mutex ) -// httpOptions tunes the HTTP transport of the tier backend's S3 client. The -// zero value keeps the SDK's default client. +// A tier endpoint that drops packets or accepts connections without answering +// must not hold a read, or a volume server's startup, for minutes per request. +var ( + s3ConnectTimeout = 10 * time.Second + s3ReadTimeout = time.Minute +) + +// httpOptions tunes the HTTP transport of the tier backend's S3 client. type httpOptions struct { // disableHTTP2 keeps the client on HTTP/1.1. Over HTTP/2 every upload and // range read to the endpoint shares one connection and its flow-control @@ -36,12 +43,10 @@ type httpOptions struct { idleConnTimeout time.Duration } -// newTransport returns the transport for the options, or nil for the default. func (o httpOptions) newTransport() *http.Transport { - if o == (httpOptions{}) { - return nil - } t := http.DefaultTransport.(*http.Transport).Clone() + t.DialContext = (&net.Dialer{Timeout: s3ConnectTimeout, KeepAlive: 30 * time.Second}).DialContext + t.ResponseHeaderTimeout = s3ReadTimeout if o.disableHTTP2 { t.ForceAttemptHTTP2 = false t.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{} @@ -73,9 +78,7 @@ func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string, Endpoint: aws.String(endpoint), S3ForcePathStyle: aws.Bool(forcePathStyle), S3DisableContentMD5Validation: aws.Bool(true), - } - if t := httpOpts.newTransport(); t != nil { - config.HTTPClient = &http.Client{Transport: t} + HTTPClient: &http.Client{Transport: httpOpts.newTransport()}, } if awsAccessKeyId != "" && awsSecretAccessKey != "" { config.Credentials = credentials.NewStaticCredentials(awsAccessKeyId, awsSecretAccessKey, "") diff --git a/weed/storage/backend/s3_backend/s3_sessions_test.go b/weed/storage/backend/s3_backend/s3_sessions_test.go index eb9f621c4d..eb5699259c 100644 --- a/weed/storage/backend/s3_backend/s3_sessions_test.go +++ b/weed/storage/backend/s3_backend/s3_sessions_test.go @@ -1,13 +1,18 @@ package s3_backend import ( + "bufio" "math" + "net" "net/http" "strconv" + "strings" + "sync" "testing" "time" "github.com/aws/aws-sdk-go/service/s3" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" ) func backendTransport(t *testing.T, s *S3BackendStorage) *http.Transport { @@ -54,17 +59,103 @@ func TestS3BackendHTTPOptionsReachTheClient(t *testing.T) { } } -func TestS3BackendDefaultHTTPClientUnchanged(t *testing.T) { +func TestS3BackendDefaultTransportHasTimeouts(t *testing.T) { s, err := newS3BackendStorage(testProperties{"endpoint": "http://127.0.0.1:2"}, "", "http-defaults") if err != nil { t.Fatal(err) } - if transport := backendTransport(t, s); transport != nil && transport != http.DefaultTransport { - t.Fatalf("no options set, yet the client got a custom transport %+v", transport) + transport := backendTransport(t, s) + if transport == nil || transport == http.DefaultTransport { + t.Fatalf("the client must not use the shared default transport, got %+v", transport) } - if (httpOptions{}).newTransport() != nil { - t.Fatal("zero options must keep the default client") + if transport.ResponseHeaderTimeout != s3ReadTimeout { + t.Errorf("ResponseHeaderTimeout = %v, want %v", transport.ResponseHeaderTimeout, s3ReadTimeout) } + if transport.DialContext == nil { + t.Error("no dialer with a connect timeout") + } + if !transport.ForceAttemptHTTP2 { + t.Error("default options must keep HTTP/2") + } +} + +// stallingEndpoint accepts connections and holds them open. With a non-empty +// head, it answers each request with head and then sends nothing more. +func stallingEndpoint(t *testing.T, head string) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + var conns []net.Conn + var mu sync.Mutex + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + mu.Lock() + conns = append(conns, c) + mu.Unlock() + if head != "" { + go func() { + if _, err := http.ReadRequest(bufio.NewReader(c)); err == nil { + c.Write([]byte(head)) + } + }() + } + } + }() + t.Cleanup(func() { + ln.Close() + mu.Lock() + defer mu.Unlock() + for _, c := range conns { + c.Close() + } + }) + return "http://" + ln.Addr().String() +} + +func readFromStallingEndpoint(t *testing.T, head, wantErr string) { + old := s3ReadTimeout + s3ReadTimeout = 100 * time.Millisecond + defer func() { s3ReadTimeout = old }() + + s, err := newS3BackendStorage(testProperties{ + "endpoint": stallingEndpoint(t, head), + "region": "us-east-1", + "bucket": "b", + "aws_access_key_id": "k", + "aws_secret_access_key": "s", + }, "", "stalling") + if err != nil { + t.Fatal(err) + } + f := s.NewStorageFile("key", &volume_server_pb.VolumeInfo{Files: []*volume_server_pb.RemoteFile{{FileSize: 1024}}}) + + done := make(chan error, 1) + go func() { + _, err := f.ReadAt(make([]byte, 8), 0) + done <- err + }() + select { + case err := <-done: + if err == nil || !strings.Contains(err.Error(), wantErr) { + t.Fatalf("want an error containing %q from a stalled endpoint, got %v", wantErr, err) + } + case <-time.After(30 * time.Second): + t.Fatal("a read from a stalled endpoint did not give up") + } +} + +func TestS3BackendReadFailsAgainstSilentEndpoint(t *testing.T) { + readFromStallingEndpoint(t, "", "timeout") +} + +func TestS3BackendReadFailsWhenBodyStalls(t *testing.T) { + readFromStallingEndpoint(t, "HTTP/1.1 206 Partial Content\r\nContent-Length: 8\r\nContent-Range: bytes 0-7/1024\r\n\r\n", "no data for") } func TestS3BackendSessionsDifferByHTTPOptions(t *testing.T) { diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go index ae2fd07c1b..6e464c0845 100644 --- a/weed/storage/volume_loading.go +++ b/weed/storage/volume_loading.go @@ -232,6 +232,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind return datFileLoadError(v.FileName(".dat"), err) } + var remoteSuperBlockErr error if alreadyHasSuperBlock { err = v.readSuperBlock() if err == nil { @@ -244,6 +245,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind if v.HasRemoteFile() { // maybe temporary network problem logRemoteSuperBlockErr(v.Id, err) + remoteSuperBlockErr = err err = nil } } else { @@ -292,7 +294,9 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind // Recover rows that deletes on a tiered read-only volume overwrote at // the front of .idx. Best effort: a volume that cannot be repaired is // still servable for everything the surviving rows index. - if restored, repairErr := v.repairIdxHeadTombstones(); repairErr != nil { + if remoteSuperBlockErr != nil { + glog.V(1).Infof("volume %d: remote backend unavailable, not recovering overwritten %s rows on this load", v.Id, v.FileName(".idx")) + } else if restored, repairErr := v.repairIdxHeadTombstones(); repairErr != nil { glog.Warningf("volume %d: recover overwritten %s rows: %v", v.Id, v.FileName(".idx"), repairErr) } else if restored > 0 { glog.V(0).Infof("volume %d: recovered %d overwritten %s rows from %s", v.Id, restored, v.FileName(".idx"), v.FileName(".dat")) diff --git a/weed/storage/volume_remote_superblock_stall_test.go b/weed/storage/volume_remote_superblock_stall_test.go new file mode 100644 index 0000000000..4aa43a7c35 --- /dev/null +++ b/weed/storage/volume_remote_superblock_stall_test.go @@ -0,0 +1,185 @@ +package storage + +import ( + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// stallingBackend serves localDirBackend's objects, but every read waits +// until release is closed: an endpoint that accepts requests and never answers. +type stallingBackend struct { + *localDirBackend + release chan struct{} + reads atomic.Int32 +} + +func (b *stallingBackend) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile { + return &stallingBackendFile{BackendStorageFile: b.localDirBackend.NewStorageFile(key, tierInfo), backend: b} +} + +type stallingBackendFile struct { + backend.BackendStorageFile + backend *stallingBackend +} + +func (f *stallingBackendFile) ReadAt(p []byte, off int64) (int, error) { + f.backend.reads.Add(1) + <-f.backend.release + return f.BackendStorageFile.ReadAt(p, off) +} + +func shortenRemoteSuperBlockWait(t *testing.T, timeout, stall time.Duration) { + t.Helper() + oldTimeout, oldStall := remoteSuperBlockTimeout, remoteSuperBlockStallPeriod + remoteSuperBlockTimeout, remoteSuperBlockStallPeriod = timeout, stall + remoteSuperBlockStalls.Delete(testBackendName) + t.Cleanup(func() { + remoteSuperBlockTimeout, remoteSuperBlockStallPeriod = oldTimeout, oldStall + remoteSuperBlockStalls.Delete(testBackendName) + }) +} + +func TestReadRemoteSuperBlockStallSkipsBackend(t *testing.T) { + shortenRemoteSuperBlockWait(t, 50*time.Millisecond, time.Hour) + b := &stallingBackend{localDirBackend: newLocalDirBackend(t), release: make(chan struct{})} + key, _, err := b.CopyFile(writeSuperBlockFile(t), nil, 0) + require.NoError(t, err) + f := b.NewStorageFile(key, &volume_server_pb.VolumeInfo{Files: []*volume_server_pb.RemoteFile{{FileSize: super_block.SuperBlockSize}}}) + + start := time.Now() + _, err = readRemoteSuperBlock(testBackendName, f) + require.Error(t, err) + require.Less(t, time.Since(start), 5*time.Second) + require.EqualValues(t, 1, b.reads.Load()) + + _, err = readRemoteSuperBlock(testBackendName, f) + require.Error(t, err) + require.EqualValues(t, 1, b.reads.Load(), "a stalled backend must not be asked again within the stall period") + + close(b.release) + remoteSuperBlockStalls.Store(testBackendName, time.Now().Add(-time.Second)) + sb, err := readRemoteSuperBlock(testBackendName, f) + require.NoError(t, err) + require.Equal(t, needle.GetCurrentVersion(), sb.Version) + _, stalled := remoteSuperBlockStalls.Load(testBackendName) + require.False(t, stalled, "a successful read clears the stall") +} + +func writeSuperBlockFile(t *testing.T) *os.File { + t.Helper() + f, err := os.Create(filepath.Join(t.TempDir(), "sb.dat")) + require.NoError(t, err) + t.Cleanup(func() { f.Close() }) + sb := super_block.SuperBlock{Version: needle.GetCurrentVersion(), ReplicaPlacement: &super_block.ReplicaPlacement{}, Ttl: &needle.TTL{}} + _, err = f.Write(sb.Bytes()) + require.NoError(t, err) + return f +} + +// A backend that never answers must not hold up the disk scan, local volumes +// included, and the tiered volumes must serve reads once it answers again. +func TestRemoteTier_DiskScanDoesNotWaitOnStalledBackend(t *testing.T) { + shortenRemoteSuperBlockWait(t, 200*time.Millisecond, time.Hour) + local := newLocalDirBackend(t) + registerTestBackend(t, local) + + dir := t.TempDir() + remoteVids := []needle.VolumeId{61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72} + for _, vid := range remoteVids { + tierUpVolume(t, dir, vid, local) + } + lv, err := NewVolume(dir, dir, "", 60, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + lv.Close() + + stalling := &stallingBackend{localDirBackend: local, release: make(chan struct{})} + backend.BackendStorages[testBackendName] = stalling + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(stalling.release) }) } + t.Cleanup(release) + + loc := &DiskLocation{ + Directory: dir, + DirectoryUuid: "test-uuid", + IdxDirectory: dir, + DiskType: types.HddType, + MaxVolumeCount: 100, + OriginalMaxVolumeCount: 100, + MinFreeSpace: util.MinFreeSpace{Type: util.AsPercent, Percent: 1, Raw: "1"}, + } + loc.volumes = make(map[needle.VolumeId]*Volume) + loc.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume) + + start := time.Now() + loc.concurrentLoadingVolumes(NeedleMapInMemory, 2, 0, 0) + require.Less(t, time.Since(start), 5*time.Second, "the disk scan waited on the stalled backend") + defer func() { + for _, v := range loc.volumes { + v.Close() + } + }() + + _, ok := loc.FindVolume(60) + require.True(t, ok, "local volume not loaded") + for _, vid := range remoteVids { + v, ok := loc.FindVolume(vid) + require.True(t, ok, "tiered volume %d not loaded", vid) + require.True(t, v.HasRemoteFile()) + } + require.Less(t, int(stalling.reads.Load()), len(remoteVids), "every tiered volume waited on the stalled backend") + + release() + v, _ := loc.FindVolume(remoteVids[0]) + _, err = v.readNeedle(newEmptyNeedle(1), nil, nil) + require.NoError(t, err, "a tiered volume must serve reads once the backend answers") +} + +// The .idx head repair reads the remote .dat, so it must wait for a load that +// can reach the backend instead of stalling this one. +func TestRemoteTier_IdxHeadRepairWaitsForReachableBackend(t *testing.T) { + shortenRemoteSuperBlockWait(t, 200*time.Millisecond, time.Hour) + local := newLocalDirBackend(t) + registerTestBackend(t, local) + + dir := t.TempDir() + const vid = needle.VolumeId(80) + tierUpVolume(t, dir, vid, local) + idxPath := filepath.Join(dir, "80.idx") + clobberIdxHead(t, idxPath, []uint64{4, 5}) + sizeBefore := fileSize(t, idxPath) + + stalling := &stallingBackend{localDirBackend: local, release: make(chan struct{})} + backend.BackendStorages[testBackendName] = stalling + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(stalling.release) }) } + time.AfterFunc(10*time.Second, release) + t.Cleanup(release) + + start := time.Now() + v := reloadVolume(t, dir, vid) + require.Less(t, time.Since(start), 5*time.Second, "the load waited on the stalled backend") + v.Close() + require.Equal(t, sizeBefore, fileSize(t, idxPath), "the repair ran against an unreachable backend") + + backend.BackendStorages[testBackendName] = local + remoteSuperBlockStalls.Delete(testBackendName) + v = reloadVolume(t, dir, vid) + defer v.Close() + require.Greater(t, fileSize(t, idxPath), sizeBefore, "the repair did not run once the backend answered") + _, err := v.readNeedle(newEmptyNeedle(1), nil, nil) + require.NoError(t, err) +} diff --git a/weed/storage/volume_super_block.go b/weed/storage/volume_super_block.go index 7035d9867c..b6faf0ea07 100644 --- a/weed/storage/volume_super_block.go +++ b/weed/storage/volume_super_block.go @@ -3,6 +3,8 @@ package storage import ( "fmt" "os" + "sync" + "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/backend" @@ -41,7 +43,11 @@ func (v *Volume) maybeWriteSuperBlock(ver needle.Version) error { } func (v *Volume) readSuperBlock() (err error) { - v.SuperBlock, err = super_block.ReadSuperBlock(v.DataBackend) + if v.HasRemoteFile() && len(v.volumeInfo.GetFiles()) > 0 { + v.SuperBlock, err = readRemoteSuperBlock(v.volumeInfo.GetFiles()[0].BackendName(), v.DataBackend) + } else { + v.SuperBlock, err = super_block.ReadSuperBlock(v.DataBackend) + } if v.volumeInfo != nil && v.volumeInfo.Replication != "" { if replication, err := super_block.NewReplicaPlacementFromString(v.volumeInfo.Replication); err != nil { return fmt.Errorf("Error parse volume %d replication %s : %v", v.Id, v.volumeInfo.Replication, err) @@ -51,3 +57,38 @@ func (v *Volume) readSuperBlock() (err error) { } return err } + +// A tiered volume loads without its superblock when the backend cannot serve +// it; once a read stalls, the backend's other volumes skip it for a while. +var ( + remoteSuperBlockTimeout = 10 * time.Second + remoteSuperBlockStallPeriod = time.Minute + remoteSuperBlockStalls sync.Map +) + +func readRemoteSuperBlock(backendName string, datBackend backend.BackendStorageFile) (super_block.SuperBlock, error) { + if until, ok := remoteSuperBlockStalls.Load(backendName); ok && time.Now().Before(until.(time.Time)) { + return super_block.SuperBlock{}, fmt.Errorf("backend %s stalled on a super block read, skipped", backendName) + } + type result struct { + superBlock super_block.SuperBlock + err error + } + done := make(chan result, 1) + go func() { + sb, err := super_block.ReadSuperBlock(datBackend) + done <- result{sb, err} + }() + timer := time.NewTimer(remoteSuperBlockTimeout) + defer timer.Stop() + select { + case r := <-done: + if r.err == nil { + remoteSuperBlockStalls.Delete(backendName) + } + return r.superBlock, r.err + case <-timer.C: + remoteSuperBlockStalls.Store(backendName, time.Now().Add(remoteSuperBlockStallPeriod)) + return super_block.SuperBlock{}, fmt.Errorf("backend %s did not return the super block within %v", backendName, remoteSuperBlockTimeout) + } +}