gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-20 11:32:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(logging): standardize protocol and observability events (#3419)

Browse Source 
* refactor(logging): standardize object capacity events

* refactor(logging): standardize protocol server events

* refactor(logging): standardize swift protocol events

* refactor(logging): standardize observability events

* refactor(logging): move masking helper and extend guardrails
This commit is contained in:
houseme
2026-06-14 07:14:45 +08:00
committed by GitHub
parent 22460243bf
commit efa89a98ed
48 changed files with 2976 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/protocols/src/sftp/config.rs
+65 -18
View File
@@ -36,6 +36,11 @@ use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use thiserror::Error;
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_CONFIG: &str = "sftp_config";
const EVENT_SFTP_CONFIG_STATE: &str = "sftp_config_state";
const EVENT_SFTP_HOST_KEY_STATE: &str = "sftp_host_key_state";
/// Upper bound on file size accepted as a candidate host key (1 MiB).
/// Guards against accidentally reading huge non-key files in the host
/// key directory. Real keys are well under 10 KiB.
@@ -203,11 +208,15 @@ impl SftpConfig {
Some(n) if (HANDLES_PER_SESSION_MIN..=HANDLES_PER_SESSION_MAX).contains(&n) => Some(n),
Some(n) => {
tracing::warn!(
event = EVENT_SFTP_CONFIG_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
requested = n,
min = HANDLES_PER_SESSION_MIN,
max = HANDLES_PER_SESSION_MAX,
default = DEFAULT_HANDLES_PER_SESSION,
"RUSTFS_SFTP_HANDLES_PER_SESSION out of range. Falling back to the default.",
result = "handles_per_session_out_of_range",
"sftp config state changed",
);
None
}
@@ -227,11 +236,15 @@ impl SftpConfig {
Some(n) if (BACKEND_OP_TIMEOUT_MIN_SECS..=BACKEND_OP_TIMEOUT_MAX_SECS).contains(&n) => Some(n),
Some(n) => {
tracing::warn!(
event = EVENT_SFTP_CONFIG_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
requested = n,
min = BACKEND_OP_TIMEOUT_MIN_SECS,
max = BACKEND_OP_TIMEOUT_MAX_SECS,
default = DEFAULT_BACKEND_OP_TIMEOUT_SECS,
"RUSTFS_SFTP_BACKEND_OP_TIMEOUT_SECS out of range. Falling back to the default.",
result = "backend_timeout_out_of_range",
"sftp config state changed",
);
None
}
@@ -254,11 +267,15 @@ impl SftpConfig {
Some(n) if (READ_CACHE_WINDOW_MIN..=READ_CACHE_WINDOW_MAX).contains(&n) => Some(n),
Some(n) => {
tracing::warn!(
event = EVENT_SFTP_CONFIG_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
requested = n,
min = READ_CACHE_WINDOW_MIN,
max = READ_CACHE_WINDOW_MAX,
default = READ_CACHE_WINDOW_DEFAULT,
"RUSTFS_SFTP_READ_CACHE_WINDOW_BYTES out of range. Set to 0 to disable the cache, or to a value between the named bounds. Falling back to the default.",
result = "read_cache_window_out_of_range",
"sftp config state changed",
);
None
}
@@ -277,10 +294,14 @@ impl SftpConfig {
Some(n) if n >= READ_CACHE_TOTAL_MEM_MIN => Some(n),
Some(n) => {
tracing::warn!(
event = EVENT_SFTP_CONFIG_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
requested = n,
min = READ_CACHE_TOTAL_MEM_MIN,
default = READ_CACHE_TOTAL_MEM_DEFAULT,
"RUSTFS_SFTP_READ_CACHE_TOTAL_MEM_BYTES below minimum. Falling back to the default.",
result = "read_cache_total_mem_below_minimum",
"sftp config state changed",
);
None
}
@@ -337,9 +358,13 @@ impl SftpConfig {
Ok(m) => m,
Err(e) => {
tracing::warn!(
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
path = %path.display(),
err = %e,
"cannot stat file, skipping"
result = "stat_failed",
"sftp host key state changed"
);
continue;
}
@@ -353,9 +378,13 @@ impl SftpConfig {
let file_size = metadata.len();
if file_size == 0 || file_size > MAX_HOST_KEY_FILE_SIZE {
tracing::debug!(
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
path = %path.display(),
size = file_size,
"skipping file: size outside valid key range"
result = "size_out_of_range",
"sftp host key state changed"
);
continue;
}
@@ -369,9 +398,13 @@ impl SftpConfig {
Ok(d) => d,
Err(e) => {
tracing::warn!(
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
path = %path.display(),
err = %e,
"cannot read file, skipping"
result = "read_failed",
"sftp host key state changed"
);
continue;
}
@@ -380,9 +413,13 @@ impl SftpConfig {
match russh::keys::decode_secret_key(&data, None) {
Ok(key) => {
tracing::info!(
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
path = %path.display(),
algorithm = ?key.algorithm(),
"loaded host key"
state = "loaded",
"sftp host key state changed"
);
keys.push(key);
}
@@ -396,15 +433,23 @@ impl SftpConfig {
// reason in the log.
if data.contains(PEM_BEGIN_MARKER) {
tracing::warn!(
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
path = %path.display(),
err = %e,
"file looks like a private key but failed to decode (passphrase-protected keys are not supported)"
result = "decode_failed",
"sftp host key state changed"
);
} else {
tracing::debug!(
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
path = %path.display(),
err = %e,
"not a valid private key, skipping"
result = "not_a_private_key",
"sftp host key state changed"
);
}
}
@@ -440,9 +485,13 @@ impl SftpConfig {
});
tracing::info!(
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
count = keys.len(),
dir = %host_key_dir.display(),
"host key loading complete"
state = "load_complete",
"sftp host key state changed"
);
Ok(keys)
@@ -467,14 +516,12 @@ fn warn_once_windows(host_key_dir: &Path) {
static WARN: Once = Once::new();
WARN.call_once(|| {
tracing::warn!(
target: "rustfs_protocols::sftp::host_key",
event = EVENT_SFTP_HOST_KEY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_CONFIG,
host_key_dir = %host_key_dir.display(),
"SFTP host key file permission enforcement is not active on \
Windows. rustfs trusts operator-managed NTFS ACLs on the \
host-key directory. Restrict read access to the running rustfs \
service account, NT AUTHORITY\\SYSTEM, and \
BUILTIN\\Administrators. Default ProgramData inheritance grants \
BUILTIN\\Users read access."
result = "windows_acl_operator_managed",
"sftp host key state changed"
);
});
}
crates/protocols/src/sftp/dir.rs
+9 -1
View File
@@ -30,6 +30,10 @@ use russh_sftp::protocol::{File, Handle, Name, StatusCode};
use rustfs_utils::path;
use s3s::dto::{ListObjectsV2Input, PutObjectInput, StreamingBlob};
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_DIR: &str = "sftp_dir";
const EVENT_SFTP_DIR_STATE: &str = "sftp_dir_state";
/// Build the conventional "." and ".." directory entries that prefix
/// the first READDIR response on every directory handle. SFTPv3 does
/// not mandate these, but POSIX clients require them. Emitting both
@@ -260,10 +264,14 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
}
if let Some(total) = truncated_at {
tracing::warn!(
event = EVENT_SFTP_DIR_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_DIR,
returned = entries.len(),
total = total,
cap = ROOT_LISTING_MAX_ENTRIES,
"root READDIR truncated: principal has more buckets than the cap",
result = "root_readdir_truncated",
"sftp dir state changed",
);
}
Ok(entries)
crates/protocols/src/sftp/driver.rs
+67 -6
View File
@@ -34,6 +34,7 @@ use crate::common::client::s3::StorageBackend;
use crate::common::gateway::{AuthorizationError, S3Action, authorize_operation};
use crate::common::session::SessionContext;
use russh_sftp::protocol::{Attrs, Data, File, FileAttributes, Handle, Name, OpenFlags, Packet, Status, StatusCode, Version};
use rustfs_utils::MaskedAccessKey;
use s3s::dto::{AbortMultipartUploadInput, CopyObjectInput, CopySource};
use std::collections::HashMap;
use std::sync::atomic::AtomicU64;
@@ -63,6 +64,12 @@ static ABORT_PERMITS: LazyLock<Arc<Semaphore>> = LazyLock::new(|| {
Arc::new(Semaphore::new(permits))
});
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_DRIVER: &str = "sftp_driver";
const EVENT_SFTP_DRIVER_STATE: &str = "sftp_driver_state";
const EVENT_SFTP_BACKEND_STATE: &str = "sftp_backend_state";
const EVENT_SFTP_AUTHZ_STATE: &str = "sftp_authz_state";
/// Per-session SFTP operation handler.
pub struct SftpDriver<S: StorageBackend + Send + Sync + 'static> {
pub(super) storage: Arc<S>,
@@ -175,9 +182,13 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
pub(super) fn enforce_server_readonly(&self) -> Result<(), SftpError> {
if self.read_only {
tracing::warn!(
event = EVENT_SFTP_DRIVER_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_DRIVER,
peer = %self.session_context.source_ip,
user = %self.session_context.principal.user_identity.credentials.access_key,
"SFTP write rejected: server is in read-only mode"
user = %MaskedAccessKey(&self.session_context.principal.user_identity.credentials.access_key),
result = "read_only_rejected",
"sftp driver state changed"
);
return Err(SftpError::code(StatusCode::PermissionDenied));
}
@@ -232,7 +243,15 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
Ok(Ok(v)) => Ok(v),
Ok(Err(e)) => Err(s3_error_to_sftp(op, e)),
Err(_elapsed) => {
tracing::warn!(op = op, timeout_secs = self.backend_op_timeout_secs, "SFTP backend operation timed out");
tracing::warn!(
event = EVENT_SFTP_BACKEND_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_DRIVER,
op = op,
timeout_secs = self.backend_op_timeout_secs,
result = "timeout",
"sftp backend state changed"
);
Err(SftpError::code(StatusCode::Failure))
}
}
@@ -251,7 +270,15 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
match tokio::time::timeout(std::time::Duration::from_secs(self.backend_op_timeout_secs), fut).await {
Ok(inner) => Ok(inner),
Err(_elapsed) => {
tracing::warn!(op = op, timeout_secs = self.backend_op_timeout_secs, "SFTP backend operation timed out");
tracing::warn!(
event = EVENT_SFTP_BACKEND_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_DRIVER,
op = op,
timeout_secs = self.backend_op_timeout_secs,
result = "timeout",
"sftp backend state changed"
);
Err(SftpError::code(StatusCode::Failure))
}
}
@@ -274,13 +301,47 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
let outcome = match tokio::time::timeout(std::time::Duration::from_secs(self.backend_op_timeout_secs), auth_fut).await {
Ok(inner) => inner,
Err(_elapsed) => {
tracing::warn!(
event = EVENT_SFTP_AUTHZ_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_DRIVER,
action = action.as_str(),
bucket = %bucket,
key = %key.unwrap_or_default(),
result = "timeout",
"sftp authz state changed"
);
return Err(auth_err_unreachable(action.as_str(), bucket, key));
}
};
match outcome {
Ok(()) => Ok(()),
Err(AuthorizationError::AccessDenied) => Err(auth_err()),
Err(AuthorizationError::IamUnavailable) => Err(auth_err_unreachable(action.as_str(), bucket, key)),
Err(AuthorizationError::AccessDenied) => {
tracing::warn!(
event = EVENT_SFTP_AUTHZ_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_DRIVER,
action = action.as_str(),
bucket = %bucket,
key = %key.unwrap_or_default(),
result = "access_denied",
"sftp authz state changed"
);
Err(auth_err())
}
Err(AuthorizationError::IamUnavailable) => {
tracing::warn!(
event = EVENT_SFTP_AUTHZ_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_DRIVER,
action = action.as_str(),
bucket = %bucket,
key = %key.unwrap_or_default(),
result = "iam_unavailable",
"sftp authz state changed"
);
Err(auth_err_unreachable(action.as_str(), bucket, key))
}
}
}
}
crates/protocols/src/sftp/errors.rs
+10 -2
View File
@@ -22,6 +22,10 @@ use russh_sftp::server::StatusReply;
use s3s::{S3Error, S3ErrorCode};
use std::{any::Any, fmt::Display};
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_ERRORS: &str = "sftp_errors";
const EVENT_SFTP_ERROR_MAPPING: &str = "sftp_error_mapping";
/// Error type for SFTP operations. Converts to StatusCode for the wire.
#[derive(Debug)]
pub struct SftpError(pub(super) StatusCode);
@@ -111,7 +115,7 @@ pub(super) fn s3_error_to_sftp<E: Display + 'static>(op: &str, err: E) -> SftpEr
BackendErrorKind::PermissionDenied => StatusCode::PermissionDenied,
BackendErrorKind::NoSuchUpload | BackendErrorKind::Other => StatusCode::Failure,
};
tracing::warn!(op = %op, err = %msg, "SFTP backend error");
tracing::warn!(event = EVENT_SFTP_ERROR_MAPPING, component = LOG_COMPONENT_PROTOCOLS, subsystem = LOG_SUBSYSTEM_SFTP_ERRORS, op = %op, err = %msg, mapped_status = ?code, "sftp error mapping");
SftpError::code(code)
}
@@ -128,10 +132,14 @@ pub(super) fn auth_err() -> SftpError {
/// deny.
pub(super) fn auth_err_unreachable(op: &str, bucket: &str, key: Option<&str>) -> SftpError {
tracing::warn!(
event = EVENT_SFTP_ERROR_MAPPING,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_ERRORS,
op = op,
bucket = %bucket,
key = key.unwrap_or("-"),
"SFTP authorisation rejected because the IAM system was unreachable"
result = "iam_unreachable",
"sftp error mapping"
);
SftpError::code(StatusCode::Failure)
}
crates/protocols/src/sftp/fallback_watchdog.rs
+8 -2
View File
@@ -29,6 +29,10 @@ use std::sync::atomic::Ordering;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio_util::sync::CancellationToken;
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_WATCHDOG: &str = "sftp_watchdog";
const EVENT_SFTP_WATCHDOG_STATE: &str = "sftp_watchdog_state";
/// Spawn a per-session silence backstop tick task.
///
/// The task holds a clone of the per-session CancellationToken and an
@@ -51,12 +55,14 @@ pub(super) fn spawn_for_session(session_diag: Arc<SessionDiag>, cancel_token: Ca
let silence_secs = silence_secs(&session_diag);
if fallback_threshold_reached(silence_secs) {
tracing::warn!(
target: "rustfs_protocols::sftp::watchdog",
event = EVENT_SFTP_WATCHDOG_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WATCHDOG,
session_id,
peer = %peer,
silence_secs,
reason = "fallback_silence",
"fallback watchdog cancelling session: silence exceeded fallback threshold",
"sftp watchdog state changed",
);
cancel_token.cancel();
break;
crates/protocols/src/sftp/server.rs
+272 -57
View File
@@ -42,6 +42,7 @@ use rustfs_config::{
DEFAULT_SFTP_HOST_KEY_RELOAD_ENABLE, DEFAULT_SFTP_HOST_KEY_RELOAD_INTERVAL, ENV_SFTP_HOST_KEY_RELOAD_ENABLE,
ENV_SFTP_HOST_KEY_RELOAD_INTERVAL,
};
use rustfs_utils::MaskedAccessKey;
use std::borrow::Cow;
use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
@@ -56,9 +57,19 @@ use tokio::sync::broadcast;
use tokio::task::JoinSet;
use tokio::time::{Duration, MissedTickBehavior, timeout};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use crate::sftp::constants::limits::SHUTDOWN_DRAIN_TIMEOUT_SECS;
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_SERVER: &str = "sftp_server";
const LOG_SUBSYSTEM_SFTP_AUTH: &str = "sftp_auth";
const EVENT_SFTP_SERVER_STATE: &str = "sftp_server_state";
const EVENT_SFTP_HOST_KEY_RELOAD_STATE: &str = "sftp_host_key_reload_state";
const EVENT_SFTP_SESSION_STATE: &str = "sftp_session_state";
const EVENT_SFTP_AUTH_STATE: &str = "sftp_auth_state";
const EVENT_SFTP_SUBSYSTEM_STATE: &str = "sftp_subsystem_state";
// Cipher, KEX, MAC, and host-key algorithm lists. All four are compile-time
// constants with no environment-variable override, so operators cannot
// accidentally downgrade to weak ciphers.
@@ -223,9 +234,13 @@ fn host_key_algorithm_rank(algorithm: keys::Algorithm) -> u8 {
fn spawn_host_key_reload_loop(config: SftpConfig, holder: Arc<SshConfigHolder>, shutdown_token: CancellationToken) {
let enabled = rustfs_utils::get_env_bool(ENV_SFTP_HOST_KEY_RELOAD_ENABLE, DEFAULT_SFTP_HOST_KEY_RELOAD_ENABLE);
if !enabled {
tracing::debug!(
"SFTP host key hot reload is disabled (set {}=1 to enable)",
ENV_SFTP_HOST_KEY_RELOAD_ENABLE
debug!(
event = EVENT_SFTP_HOST_KEY_RELOAD_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "disabled",
env_var = ENV_SFTP_HOST_KEY_RELOAD_ENABLE,
"sftp host key reload state changed"
);
return;
}
@@ -233,10 +248,14 @@ fn spawn_host_key_reload_loop(config: SftpConfig, holder: Arc<SshConfigHolder>,
let interval_secs =
rustfs_utils::get_env_u64(ENV_SFTP_HOST_KEY_RELOAD_INTERVAL, DEFAULT_SFTP_HOST_KEY_RELOAD_INTERVAL).max(5);
tracing::info!(
info!(
event = EVENT_SFTP_HOST_KEY_RELOAD_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "enabled",
host_key_dir = %config.host_key_dir.display(),
interval_secs,
"SFTP host key hot reload enabled"
"sftp host key reload state changed"
);
tokio::spawn(async move {
@@ -246,7 +265,14 @@ fn spawn_host_key_reload_loop(config: SftpConfig, holder: Arc<SshConfigHolder>,
loop {
tokio::select! {
_ = shutdown_token.cancelled() => {
tracing::info!(host_key_dir = %config.host_key_dir.display(), "SFTP host key hot reload task stopped");
info!(
event = EVENT_SFTP_HOST_KEY_RELOAD_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "stopped",
host_key_dir = %config.host_key_dir.display(),
"sftp host key reload state changed"
);
break;
}
_ = interval.tick() => {}
@@ -254,23 +280,35 @@ fn spawn_host_key_reload_loop(config: SftpConfig, holder: Arc<SshConfigHolder>,
match holder.reload_from_config(&config).await {
Ok(Some(host_key_count)) => {
tracing::info!(
info!(
event = EVENT_SFTP_HOST_KEY_RELOAD_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "reloaded",
host_key_dir = %config.host_key_dir.display(),
host_key_count,
"SFTP host keys reloaded successfully"
"sftp host key reload state changed"
);
}
Ok(None) => {
tracing::debug!(
debug!(
event = EVENT_SFTP_HOST_KEY_RELOAD_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "unchanged",
host_key_dir = %config.host_key_dir.display(),
"SFTP host key material unchanged; skipping reload"
"sftp host key reload state changed"
);
}
Err(err) => {
tracing::warn!(
warn!(
event = EVENT_SFTP_HOST_KEY_RELOAD_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "reload_failed",
host_key_dir = %config.host_key_dir.display(),
err = %err,
"SFTP host key reload failed; keeping previous keys"
"sftp host key reload state changed"
);
}
}
@@ -353,7 +391,15 @@ where
let listener = TcpListener::bind(self.config.bind_addr)
.await
.map_err(|e| SftpInitError::Server(format!("failed to bind {}: {}", self.config.bind_addr, e)))?;
tracing::info!(bind_addr = %self.config.bind_addr, "SFTP server listening");
info!(
event = EVENT_SFTP_SERVER_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "listening",
bind_addr = %self.config.bind_addr,
read_only = self.config.read_only,
"sftp server state changed"
);
let mut sessions: JoinSet<()> = JoinSet::new();
// Parent cancellation token for the lifetime of this listener.
@@ -383,9 +429,13 @@ where
self.handle_accept(accept_result, &mut sessions, &server_shutdown_token);
}
_ = shutdown_rx.recv() => {
tracing::info!(
info!(
event = EVENT_SFTP_SERVER_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "shutdown_requested",
live_sessions = sessions.len(),
"SFTP server received shutdown signal",
"sftp server state changed",
);
// Cascade cancellation to every live session and
// its watchdog before the drain loop runs. Wedged
@@ -412,11 +462,34 @@ where
while let Some(res) = sessions.try_join_next() {
let live = sessions.len();
match res {
Ok(()) => tracing::debug!(live_sessions = live, "SFTP session task finished"),
Ok(()) => debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "finished",
live_sessions = live,
"sftp session state changed"
),
Err(e) if e.is_panic() => {
tracing::error!(err = %e, live_sessions = live, "SFTP session task panicked")
error!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "task_panicked",
err = %e,
live_sessions = live,
"sftp session state changed"
)
}
Err(e) => tracing::debug!(err = %e, live_sessions = live, "SFTP session task cancelled"),
Err(e) => debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "cancelled",
err = %e,
live_sessions = live,
"sftp session state changed"
),
}
}
}
@@ -434,7 +507,14 @@ where
let (stream, peer_addr) = match accept_result {
Ok(v) => v,
Err(e) => {
tracing::warn!(err = %e, "failed to accept connection");
warn!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "accept_failed",
err = %e,
"sftp session state changed"
);
return;
}
};
@@ -454,7 +534,13 @@ where
// is still consistent across panics, and the accept loop
// must keep running.
let mut reg = self.session_registry.lock().unwrap_or_else(|poisoned| {
tracing::warn!("session registry mutex poisoned, recovering");
warn!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "registry_poisoned",
"sftp session state changed"
);
poisoned.into_inner()
});
reg.push(Arc::downgrade(&session_diag));
@@ -472,10 +558,14 @@ where
let watchdog_socket = {
let socket = wedge_watchdog::dup_socket(&stream);
if socket.is_none() {
tracing::warn!(
warn!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "watchdog_socket_unavailable",
peer = %peer_addr,
session_id = session_diag.session_id,
"wedge watchdog: dup_socket failed, session has no wedge protection (rare; usually fd exhaustion)",
"sftp session state changed",
);
}
socket
@@ -500,13 +590,17 @@ where
session_diag: handler_session_diag,
};
tracing::debug!(
debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "spawned",
peer = %peer_addr,
// sessions.len() reads pre-spawn, so add one to include
// the session about to be inserted.
live_sessions = sessions.len() + 1,
session_id = session_diag.session_id,
"SFTP accept: spawning session task",
"sftp session state changed",
);
#[cfg(target_os = "linux")]
sessions.spawn(run_session(
@@ -670,7 +764,14 @@ async fn run_session<S>(
) where
S: StorageBackend + Send + Sync + 'static,
{
tracing::debug!(peer = %peer_addr, "SFTP session task entered");
debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "task_entered",
peer = %peer_addr,
"sftp session state changed"
);
// run_stream covers SSH KEX and password auth. Cap with a
// wallclock deadline so a peer that completes TCP but stalls
// before KEXINIT (or that drives KEX or auth so slowly that no
@@ -681,19 +782,38 @@ async fn run_session<S>(
let session = match timeout(handshake_deadline, russh::server::run_stream(ssh_config, stream, handler)).await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
tracing::debug!(peer = %peer_addr, err = %e, "SSH session setup failed");
debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "setup_failed",
peer = %peer_addr,
err = %e,
"sftp session state changed"
);
return;
}
Err(_elapsed) => {
tracing::warn!(
warn!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "handshake_timeout",
peer = %peer_addr,
deadline_secs = HANDSHAKE_DEADLINE_SECS,
"SSH handshake exceeded deadline; dropping connection",
"sftp session state changed",
);
return;
}
};
tracing::debug!(peer = %peer_addr, "SFTP session run_stream returned; awaiting session loop");
debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "handshake_completed",
peer = %peer_addr,
"sftp session state changed"
);
// Spawn the per-session watchdog. On Linux the wedge watchdog
// observes the SFTP-handler activity stamp and the kernel TCP
// state. On wedge detection it shuts the duplicated socket down
@@ -722,16 +842,38 @@ async fn run_session<S>(
let session_result = match session_result {
Some(r) => r,
None => {
tracing::warn!(peer = %peer_addr, "SFTP session cancelled (watchdog or server shutdown)");
warn!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "cancelled",
peer = %peer_addr,
"sftp session state changed"
);
return;
}
};
match session_result {
Ok(()) => {
tracing::debug!(peer = %peer_addr, "SFTP session ended cleanly");
debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "ended_cleanly",
peer = %peer_addr,
"sftp session state changed"
);
}
Err(e) => {
tracing::debug!(peer = %peer_addr, err = %e, "SSH session ended with error");
debug!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "ended_with_error",
peer = %peer_addr,
err = %e,
"sftp session state changed"
);
}
}
}
@@ -757,16 +899,33 @@ async fn drain_sessions(mut sessions: JoinSet<()>) {
if let Err(e) = res
&& e.is_panic()
{
tracing::error!(err = %e, "SFTP session task panicked during drain");
error!(
event = EVENT_SFTP_SESSION_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "drain_task_panicked",
err = %e,
"sftp session state changed"
);
}
}
};
match timeout(Duration::from_secs(SHUTDOWN_DRAIN_TIMEOUT_SECS), drain).await {
Ok(()) => tracing::info!("SFTP session drain complete"),
Err(_) => tracing::warn!(
Ok(()) => info!(
event = EVENT_SFTP_SERVER_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "drain_completed",
"sftp server state changed"
),
Err(_) => warn!(
event = EVENT_SFTP_SERVER_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "drain_timeout",
timeout_secs = SHUTDOWN_DRAIN_TIMEOUT_SECS,
live = sessions.len(),
"SFTP session drain timed out, cancelling remaining sessions",
"sftp server state changed",
),
}
}
@@ -872,6 +1031,7 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
) -> impl std::future::Future<Output = Result<Auth, Self::Error>> + Send {
let user = user.to_owned();
let password = password.to_owned();
let masked_user = MaskedAccessKey(&user).to_string();
let peer_addr = self.peer_addr;
let session_diag = Arc::clone(&self.session_diag);
session_diag.stamp();
@@ -880,7 +1040,15 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let iam_sys = match rustfs_iam::get() {
Ok(sys) => sys,
Err(e) => {
tracing::error!(err = %e, "IAM system unavailable");
error!(
event = EVENT_SFTP_AUTH_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_AUTH,
result = "iam_unavailable",
peer = %peer_addr,
err = %e,
"sftp auth state changed"
);
return Ok(Auth::reject());
}
};
@@ -888,10 +1056,15 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let (identity_opt, is_valid) = match iam_sys.check_key(&user).await {
Ok(result) => result,
Err(e) => {
tracing::error!(
user = %user,
error!(
event = EVENT_SFTP_AUTH_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_AUTH,
result = "check_key_failed",
user = %masked_user,
peer = %peer_addr,
err = %e,
"IAM check_key error"
"sftp auth state changed"
);
return Ok(Auth::reject());
}
@@ -900,10 +1073,14 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let identity = match identity_opt {
Some(id) => id,
None => {
tracing::warn!(
user = %user,
warn!(
event = EVENT_SFTP_AUTH_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_AUTH,
result = "unknown_access_key",
user = %masked_user,
peer = %peer_addr,
"SFTP auth rejected: unknown access key"
"sftp auth state changed"
);
return Ok(Auth::reject());
}
@@ -912,10 +1089,14 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
// Reject disabled or expired accounts. FTPS checks this at
// ftps/server.rs:286. SFTP must do the same.
if !is_valid {
tracing::warn!(
user = %user,
warn!(
event = EVENT_SFTP_AUTH_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_AUTH,
result = "account_inactive",
user = %masked_user,
peer = %peer_addr,
"SFTP auth rejected: account disabled or expired"
"sftp auth state changed"
);
return Ok(Auth::reject());
}
@@ -926,10 +1107,14 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let secret_matches: bool = identity.credentials.secret_key.as_bytes().ct_eq(password.as_bytes()).into();
if !secret_matches {
tracing::warn!(
user = %user,
warn!(
event = EVENT_SFTP_AUTH_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_AUTH,
result = "invalid_secret_key",
user = %masked_user,
peer = %peer_addr,
"SFTP auth rejected: invalid secret key"
"sftp auth state changed"
);
return Ok(Auth::reject());
}
@@ -937,10 +1122,14 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let principal = ProtocolPrincipal::new(Arc::new(identity));
self.session_context = Some(SessionContext::new(principal, Protocol::Sftp, peer_addr.ip()));
tracing::info!(
user = %user,
debug!(
event = EVENT_SFTP_AUTH_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_AUTH,
result = "authenticated",
user = %masked_user,
peer = %peer_addr,
"SFTP auth accepted"
"sftp auth state changed"
);
Ok(Auth::Accept)
}
@@ -992,10 +1181,14 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let inputs: Result<Option<RunInputs<S>>, Self::Error> = (|| {
if name != SFTP_SUBSYSTEM_NAME {
tracing::warn!(
warn!(
event = EVENT_SFTP_SUBSYSTEM_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "unsupported_subsystem",
subsystem = %name,
peer = %self.peer_addr,
"rejecting unsupported subsystem"
"sftp subsystem state changed"
);
session.channel_failure(channel_id)?;
return Ok(None);
@@ -1004,9 +1197,14 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let channel = match self.channels.remove(&channel_id) {
Some(ch) => ch,
None => {
tracing::error!(
error!(
event = EVENT_SFTP_SUBSYSTEM_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "channel_missing",
channel = ?channel_id,
"subsystem_request: no channel found"
peer = %self.peer_addr,
"sftp subsystem state changed"
);
session.channel_failure(channel_id)?;
return Ok(None);
@@ -1016,13 +1214,30 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
let session_context = match self.session_context.clone() {
Some(ctx) => ctx,
None => {
tracing::error!("subsystem_request before authentication");
error!(
event = EVENT_SFTP_SUBSYSTEM_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
result = "unauthenticated_request",
channel = ?channel_id,
peer = %self.peer_addr,
"sftp subsystem state changed"
);
session.channel_failure(channel_id)?;
return Ok(None);
}
};
session.channel_success(channel_id)?;
debug!(
event = EVENT_SFTP_SUBSYSTEM_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_SERVER,
state = "accepted",
channel = ?channel_id,
peer = %self.peer_addr,
"sftp subsystem state changed"
);
Ok(Some(RunInputs {
stream: channel.into_stream(),
crates/protocols/src/sftp/wedge_watchdog.rs
+8 -2
View File
@@ -62,6 +62,10 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::net::TcpStream;
use tokio_util::sync::CancellationToken;
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_WATCHDOG: &str = "sftp_watchdog";
const EVENT_SFTP_WATCHDOG_STATE: &str = "sftp_watchdog_state";
/// Reason a watchdog cancelled its session. Included in the warn log
/// the watchdog writes at cancel time so operators can correlate the
/// cancel with the upstream client behaviour.
@@ -140,12 +144,14 @@ pub(super) fn spawn_for_session(session_diag: Arc<SessionDiag>, socket: Socket,
}
Decision::Cancel(reason) => {
tracing::warn!(
target: "rustfs_protocols::sftp::watchdog",
event = EVENT_SFTP_WATCHDOG_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WATCHDOG,
session_id,
peer = %peer,
silence_secs,
reason = reason.as_str(),
"wedge watchdog cancelling session: russh select! parked outside its arms",
"sftp watchdog state changed",
);
cancel_token.cancel();
break;
crates/protocols/src/sftp/write.rs
+110 -18
View File
@@ -32,11 +32,19 @@ use crate::common::gateway::S3Action;
use bytes::Bytes;
use futures_util::stream;
use russh_sftp::protocol::{FileAttributes, Handle, OpenFlags, StatusCode};
use rustfs_utils::MaskedAccessKey;
use s3s::dto::{
AbortMultipartUploadInput, CompleteMultipartUploadInput, CompletedMultipartUpload, CompletedPart as S3CompletedPart,
CopySource, CreateMultipartUploadInput, PutObjectInput, StreamingBlob, UploadPartCopyInput, UploadPartInput,
};
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
const LOG_SUBSYSTEM_SFTP_WRITE: &str = "sftp_write";
const EVENT_SFTP_WRITE_STATE: &str = "sftp_write_state";
const EVENT_SFTP_MULTIPART_STATE: &str = "sftp_multipart_state";
const EVENT_SFTP_ABORT_STATE: &str = "sftp_abort_state";
const EVENT_SFTP_MULTIPART_COPY_STATE: &str = "sftp_multipart_copy_state";
/// Running byte count for a write handle's current phase. Buffering is
/// part_buffer.len(). Streaming is (parts_done * part_size) +
/// part_buffer.len(). Failed is treated as zero in saturating mode and
@@ -62,7 +70,13 @@ pub(super) fn write_dispatch_byte_count(phase: &WritePhase, part_size: u64, satu
(Some(value), _) => Ok(value),
(None, true) => Ok(u64::MAX),
(None, false) => {
tracing::warn!("SFTP write running total overflowed u64");
tracing::warn!(
event = EVENT_SFTP_WRITE_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
result = "byte_count_overflow",
"sftp write state changed"
);
Err(SftpError::code(StatusCode::Failure))
}
}
@@ -71,7 +85,13 @@ pub(super) fn write_dispatch_byte_count(phase: &WritePhase, part_size: u64, satu
if saturating {
Ok(0)
} else {
tracing::warn!("SFTP write rejected: handle already poisoned by earlier upload failure");
tracing::warn!(
event = EVENT_SFTP_WRITE_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
result = "handle_poisoned",
"sftp write state changed"
);
Err(SftpError::code(StatusCode::Failure))
}
}
@@ -90,7 +110,13 @@ pub(super) fn write_dispatch_append_bytes(phase: &mut WritePhase, data: &[u8]) -
Ok(())
}
WritePhase::Failed { .. } => {
tracing::error!("SFTP write_dispatch_append_bytes reached a Failed handle, internal invariant broken");
tracing::error!(
event = EVENT_SFTP_WRITE_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
result = "append_on_failed_handle",
"sftp write state changed"
);
Err(SftpError::code(StatusCode::Failure))
}
}
@@ -335,10 +361,14 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
if attempt > 0 {
tokio::time::sleep(std::time::Duration::from_millis(COMMIT_WRITE_BACKOFF_MS[attempt - 1])).await;
tracing::warn!(
event = EVENT_SFTP_WRITE_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %sanitise_control_bytes(bucket),
key = %sanitise_control_bytes(key),
attempt = attempt,
"retrying commit_write put_object after retryable backend error",
state = "retrying_put_object",
"sftp write state changed",
);
}
@@ -378,9 +408,13 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
// that proof, log and surface Failure rather than panicking
// the session task.
tracing::error!(
event = EVENT_SFTP_WRITE_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %sanitise_control_bytes(bucket),
key = %sanitise_control_bytes(key),
"commit_write retry loop fell through without returning",
result = "retry_loop_fell_through",
"sftp write state changed",
);
Err(SftpError::code(StatusCode::Failure))
}
@@ -418,7 +452,15 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
.await?;
let e_tag = out.e_tag.ok_or_else(|| {
tracing::warn!(upload_id = %upload_id, part_number = part_number, "UploadPart returned no ETag");
tracing::warn!(
event = EVENT_SFTP_MULTIPART_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
upload_id = %upload_id,
part_number = part_number,
result = "etag_missing",
"sftp multipart state changed"
);
SftpError::code(StatusCode::Failure)
})?;
@@ -495,9 +537,13 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
let upload_id = out.upload_id.ok_or_else(|| {
tracing::warn!(
event = EVENT_SFTP_MULTIPART_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %sanitise_control_bytes(bucket),
key = %sanitise_control_bytes(key),
"CreateMultipartUpload returned no upload_id"
result = "upload_id_missing",
"sftp multipart state changed"
);
SftpError::code(StatusCode::Failure)
})?;
@@ -595,7 +641,15 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
let current_len = write_dispatch_byte_count(phase, part_size, false)?;
if offset != current_len {
tracing::warn!(offset = offset, buffered = current_len, "SFTP write rejected: non-sequential offset");
tracing::warn!(
event = EVENT_SFTP_WRITE_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
offset,
buffered = current_len,
result = "non_sequential_offset",
"sftp write state changed"
);
return Err(SftpError::code(StatusCode::Failure));
}
@@ -650,7 +704,11 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
WritePhase::Buffering { part_buffer } => std::mem::take(part_buffer),
_ => {
tracing::error!(
"SFTP write_dispatch_begin_streaming lost Buffering phase between check and extract, internal invariant broken"
event = EVENT_SFTP_MULTIPART_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
result = "buffering_phase_lost",
"sftp multipart state changed"
);
return Err(SftpError::code(StatusCode::Failure));
}
@@ -707,10 +765,14 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
} => {
if *next_part_number > S3_MAX_MULTIPART_PARTS {
tracing::warn!(
event = EVENT_SFTP_MULTIPART_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %bucket,
key = %key,
limit = S3_MAX_MULTIPART_PARTS,
"SFTP write would exceed the S3 multipart parts limit",
result = "parts_limit_exceeded",
"sftp multipart state changed",
);
let upload_id_for_fail = upload_id.clone();
let abort_authorized_for_fail = *abort_authorized;
@@ -724,7 +786,13 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
(upload_id.clone(), *abort_authorized, *next_part_number, drained)
}
_ => {
tracing::error!("SFTP write_dispatch_flush_one_part called without Streaming phase, internal invariant broken");
tracing::error!(
event = EVENT_SFTP_MULTIPART_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
result = "flush_without_streaming",
"sftp multipart state changed"
);
return Err(SftpError::code(StatusCode::Failure));
}
};
@@ -745,7 +813,11 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
}
_ => {
tracing::error!(
"SFTP write_dispatch_flush_one_part post-upload arm without Streaming phase, internal invariant broken"
event = EVENT_SFTP_MULTIPART_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
result = "post_upload_phase_missing",
"sftp multipart state changed"
);
Err(SftpError::code(StatusCode::Failure))
}
@@ -806,22 +878,30 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
if abort_authorized {
if let Err(abort_err) = self.abort_upload_with_auth(bucket, key, upload_id).await {
tracing::warn!(
event = EVENT_SFTP_ABORT_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %bucket,
key = %key,
upload_id = %upload_id,
err = ?abort_err,
"abort after {context} also failed; S3 lifecycle must clean up",
context = context,
result = "abort_failed",
"sftp abort state changed",
);
}
} else {
tracing::warn!(
event = EVENT_SFTP_ABORT_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %bucket,
key = %key,
upload_id = %upload_id,
access_key = %self.access_key(),
"skipped abort at close ({context}): principal lacks s3:AbortMultipartUpload, bucket lifecycle rules must reclaim parts",
access_key = %MaskedAccessKey(self.access_key()),
context = context,
result = "abort_skipped_unauthorized",
"sftp abort state changed",
);
}
}
@@ -871,11 +951,15 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
if !part_buffer.is_empty() {
if next_part_number > S3_MAX_MULTIPART_PARTS {
tracing::warn!(
event = EVENT_SFTP_MULTIPART_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %bucket,
key = %key,
upload_id = %upload_id,
limit = S3_MAX_MULTIPART_PARTS,
"SFTP close rejected: trailing part would exceed S3 multipart parts limit",
result = "final_part_limit_exceeded",
"sftp multipart state changed",
);
self.close_abort_or_skip(bucket, key, &upload_id, abort_authorized, "parts-limit breach")
.await;
@@ -932,11 +1016,15 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
let target = needed.max(configured).max(S3_MIN_PART_SIZE);
if target > S3_MAX_PART_SIZE {
tracing::warn!(
event = EVENT_SFTP_MULTIPART_COPY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
bucket = %dst_bucket,
key = %sanitise_control_bytes(dst_key),
content_length,
target,
"multipart copy refused: per-part size exceeds S3_MAX_PART_SIZE"
result = "part_size_exceeded",
"sftp multipart copy state changed"
);
return Err(SftpError::code(StatusCode::Failure));
}
@@ -986,9 +1074,13 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
let e_tag = out.copy_part_result.and_then(|r| r.e_tag).ok_or_else(|| {
tracing::warn!(
event = EVENT_SFTP_MULTIPART_COPY_STATE,
component = LOG_COMPONENT_PROTOCOLS,
subsystem = LOG_SUBSYSTEM_SFTP_WRITE,
upload_id = %mp.upload_id,
part_number = part_number,
"UploadPartCopy returned no ETag"
result = "etag_missing",
"sftp multipart copy state changed"
);
SftpError::code(StatusCode::Failure)
})?;