mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-26 05:56:50 +00:00
fix(logging): enforce single-writer sinks and bound tracing (#4765)
* fix(obs): prevent rolling log stdout aliasing * fix(ecstore): bound hot-path tracing payloads * test(logging): guard service and disk log invariants * fix(obs): silence useless_conversion on st_dev for Linux clippy rustix's Stat.st_dev is u64 on Linux/glibc, making u64::try_from a no-op that trips clippy::useless_conversion under -D warnings. The conversion is still needed on macOS/BSD where st_dev is a signed dev_t, so suppress the lint on that line rather than dropping the portable fallible conversion. * fix(logging): keep stdout sink validation portable (#4769) * fix(obs): keep stdout device conversion portable * docs(logging): update single-writer plan status --------- Co-authored-by: overtrue <anzhengchao@gmail.com> Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
@@ -68,7 +68,7 @@ use tokio::{
|
||||
};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tonic::{Code, Request, service::interceptor::InterceptedService, transport::Channel};
|
||||
use tracing::{debug, warn};
|
||||
use tracing::{debug, trace, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
@@ -1151,40 +1151,40 @@ fn decode_batch_read_version_response_items(
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl DiskAPI for RemoteDisk {
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn to_string(&self) -> String {
|
||||
self.endpoint.to_string()
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn is_online(&self) -> bool {
|
||||
// If disk is marked as faulty, consider it offline
|
||||
!self.health.is_faulty()
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn is_local(&self) -> bool {
|
||||
false
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn host_name(&self) -> String {
|
||||
self.endpoint.host_port()
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn endpoint(&self) -> Endpoint {
|
||||
self.endpoint.clone()
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn close(&self) -> Result<()> {
|
||||
self.cancel_token.cancel();
|
||||
Ok(())
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn get_disk_id(&self) -> Result<Option<Uuid>> {
|
||||
Ok(*self.id.lock().await)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn set_disk_id(&self, id: Option<Uuid>) -> Result<()> {
|
||||
let mut lock = self.id.lock().await;
|
||||
*lock = id;
|
||||
@@ -1192,12 +1192,12 @@ impl DiskAPI for RemoteDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn path(&self) -> PathBuf {
|
||||
PathBuf::from(self.endpoint.get_file_path())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn get_disk_location(&self) -> DiskLocation {
|
||||
DiskLocation {
|
||||
pool_idx: {
|
||||
@@ -1224,9 +1224,9 @@ impl DiskAPI for RemoteDisk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn make_volume(&self, volume: &str) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1261,9 +1261,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1298,9 +1298,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1339,9 +1339,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1378,9 +1378,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_volume(&self, volume: &str, force_delete: bool) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1416,7 +1416,7 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_version(
|
||||
&self,
|
||||
volume: &str,
|
||||
@@ -1425,7 +1425,7 @@ impl DiskAPI for RemoteDisk {
|
||||
force_del_marker: bool,
|
||||
opts: DeleteOptions,
|
||||
) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1476,9 +1476,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_versions(&self, volume: &str, versions: Vec<FileInfoVersions>, opts: DeleteOptions) -> Vec<Option<Error>> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1606,9 +1606,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1646,9 +1646,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1722,9 +1722,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1771,7 +1771,7 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_version(
|
||||
&self,
|
||||
_org_volume: &str,
|
||||
@@ -1780,7 +1780,7 @@ impl DiskAPI for RemoteDisk {
|
||||
version_id: &str,
|
||||
opts: &ReadOptions,
|
||||
) -> Result<FileInfo> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1833,7 +1833,7 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self, req))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn batch_read_version(&self, req: BatchReadVersionReq) -> Result<Vec<BatchReadVersionResp>> {
|
||||
validate_batch_read_version_item_count(req.items.len())?;
|
||||
|
||||
@@ -1844,7 +1844,7 @@ impl DiskAPI for RemoteDisk {
|
||||
}
|
||||
record_batch_read_version_gate_decision(mode, BATCH_READ_VERSION_GATE_ATTEMPT);
|
||||
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1926,9 +1926,9 @@ impl DiskAPI for RemoteDisk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -1971,7 +1971,7 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_data(
|
||||
&self,
|
||||
src_volume: &str,
|
||||
@@ -1980,7 +1980,7 @@ impl DiskAPI for RemoteDisk {
|
||||
dst_volume: &str,
|
||||
dst_path: &str,
|
||||
) -> Result<RenameDataResp> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2032,9 +2032,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn list_dir(&self, _origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>> {
|
||||
debug!("list_dir {}/{}", volume, dir_path);
|
||||
trace!(volume, dir_path, "Remote disk list_dir RPC started");
|
||||
|
||||
self.execute_with_timeout(
|
||||
|| async {
|
||||
@@ -2064,9 +2064,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self, wr))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2142,12 +2142,12 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
|
||||
self.read_file_stream(volume, path, 0, 0).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
|
||||
// warn!(
|
||||
// "disk remote read_file_stream {}/{}/{} offset={} length={}",
|
||||
@@ -2177,7 +2177,7 @@ impl DiskAPI for RemoteDisk {
|
||||
|
||||
/// Buffered read for remote disks.
|
||||
/// The transport stream is collected into owned Bytes for caller sharing.
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
||||
// For remote disks, use the regular reader and read into Bytes
|
||||
let reader = self.read_file_stream(volume, path, offset, length).await?;
|
||||
@@ -2192,9 +2192,9 @@ impl DiskAPI for RemoteDisk {
|
||||
Ok(Bytes::from(buffer))
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2221,7 +2221,7 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result<FileWriter> {
|
||||
// warn!(
|
||||
// "disk remote create_file {}/{}/{} file_size={}",
|
||||
@@ -2246,9 +2246,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2289,9 +2289,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2333,9 +2333,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2376,9 +2376,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2419,9 +2419,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_parts(&self, bucket: &str, paths: &[String]) -> Result<Vec<ObjectPartInfo>> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2458,9 +2458,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2501,9 +2501,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2549,9 +2549,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2602,9 +2602,9 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
|
||||
debug!(
|
||||
trace!(
|
||||
event = EVENT_REMOTE_DISK_RPC,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
|
||||
@@ -2655,7 +2655,7 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo> {
|
||||
// disk_info is idempotent/read-only, so it is eligible for the P3-3 bounded retry.
|
||||
self.execute_read_with_retry(
|
||||
@@ -2686,7 +2686,7 @@ impl DiskAPI for RemoteDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn start_scan(&self) -> ScanGuard {
|
||||
self.scanning.fetch_add(1, Ordering::Relaxed);
|
||||
ScanGuard(Arc::clone(&self.scanning))
|
||||
|
||||
@@ -4139,7 +4139,7 @@ impl LocalDisk {
|
||||
true
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn check_format_json(&self) -> Result<Metadata> {
|
||||
let md = fs::metadata(&self.format_path).await.map_err(to_unformatted_disk_error)?;
|
||||
Ok(md)
|
||||
@@ -4376,7 +4376,7 @@ impl LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
#[async_recursion::async_recursion]
|
||||
async fn delete_file(
|
||||
&self,
|
||||
@@ -4455,7 +4455,7 @@ impl LocalDisk {
|
||||
}
|
||||
|
||||
/// read xl.meta raw data
|
||||
#[tracing::instrument(level = "debug", skip(self, volume_dir, file_path))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_raw(
|
||||
&self,
|
||||
bucket: &str,
|
||||
@@ -4549,7 +4549,7 @@ impl LocalDisk {
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self, volume_dir, file_path))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_all_data_with_dmtime(
|
||||
&self,
|
||||
volume: &str,
|
||||
@@ -4874,7 +4874,7 @@ impl LocalDisk {
|
||||
}
|
||||
|
||||
// write_all_private with check_path_length
|
||||
#[tracing::instrument(level = "debug", skip_all)]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn write_all_private(&self, volume: &str, path: &str, buf: Bytes, sync: SyncMode, skip_parent: &Path) -> Result<()> {
|
||||
let file_path = self.get_object_path(volume, path)?;
|
||||
check_path_length(file_path.to_string_lossy().as_ref())?;
|
||||
@@ -5851,7 +5851,7 @@ impl DiskAPI for LocalDisk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn get_disk_id(&self) -> Result<Option<Uuid>> {
|
||||
let format_info = {
|
||||
let format_info = self.format_info.read().await;
|
||||
@@ -5931,7 +5931,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
|
||||
crate::hp_guard!("LocalDisk::read_all");
|
||||
if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE {
|
||||
@@ -5948,13 +5948,13 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip_all)]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::write_all");
|
||||
self.write_all_public(volume, path, data).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::delete");
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
@@ -5980,7 +5980,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
if !skip_access_checks(volume)
|
||||
@@ -6059,7 +6059,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(resp)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_parts(&self, bucket: &str, paths: &[String]) -> Result<Vec<ObjectPartInfo>> {
|
||||
let volume_dir = self.get_bucket_path(bucket)?;
|
||||
|
||||
@@ -6125,7 +6125,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
Ok(ret)
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
let file_path = self.get_object_path(volume, path)?;
|
||||
@@ -6211,7 +6211,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(resp)
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()> {
|
||||
let src_volume_dir = self.get_bucket_path(src_volume)?;
|
||||
let dst_volume_dir = self.get_bucket_path(dst_volume)?;
|
||||
@@ -6334,7 +6334,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::rename_file");
|
||||
let src_volume_dir = self.get_bucket_path(src_volume)?;
|
||||
@@ -6406,7 +6406,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: i64) -> Result<FileWriter> {
|
||||
crate::hp_guard!("LocalDisk::create_file");
|
||||
if !origvolume.is_empty() {
|
||||
@@ -6423,19 +6423,19 @@ impl DiskAPI for LocalDisk {
|
||||
.await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
// async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result<File> {
|
||||
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
|
||||
self.io_backend.open_write(volume, path, WriteMode::Append).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
|
||||
crate::hp_guard!("LocalDisk::read_file");
|
||||
self.io_backend.open_full_read(volume, path).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
|
||||
crate::hp_guard!("LocalDisk::read_file_stream");
|
||||
self.io_backend.open_read_stream(volume, path, offset, length).await
|
||||
@@ -6445,14 +6445,14 @@ impl DiskAPI for LocalDisk {
|
||||
// SAFETY: Unix unsafe calls in this function only query page size and mmap
|
||||
// a read-only file region after bounds and alignment are validated.
|
||||
#[allow(unsafe_code)]
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
||||
self.read_file_mmap_copy_with_metrics(volume, path, offset, length, None)
|
||||
.await
|
||||
}
|
||||
|
||||
/// File read using mmap-then-copy on Unix or efficient read on non-Unix.
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file_mmap_copy_with_metrics(
|
||||
&self,
|
||||
volume: &str,
|
||||
@@ -6464,7 +6464,7 @@ impl DiskAPI for LocalDisk {
|
||||
self.io_backend.pread_bytes(volume, path, offset, length, metrics).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>> {
|
||||
if !origvolume.is_empty() {
|
||||
let origvolume_dir = self.get_bucket_path(origvolume)?;
|
||||
@@ -6496,7 +6496,7 @@ impl DiskAPI for LocalDisk {
|
||||
}
|
||||
|
||||
// FIXME: TODO: io.writer TODO cancel
|
||||
#[tracing::instrument(level = "debug", skip(self, wr))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
|
||||
self.wait_for_startup_cleanup().await;
|
||||
|
||||
@@ -6593,7 +6593,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self, fi))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_data(
|
||||
&self,
|
||||
src_volume: &str,
|
||||
@@ -7208,7 +7208,7 @@ impl DiskAPI for LocalDisk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
|
||||
for vol in volumes {
|
||||
if let Err(e) = self.make_volume(vol).await
|
||||
@@ -7230,7 +7230,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn make_volume(&self, volume: &str) -> Result<()> {
|
||||
if !Self::is_valid_volname(volume) {
|
||||
return Err(Error::other("Invalid arguments specified"));
|
||||
@@ -7258,7 +7258,7 @@ impl DiskAPI for LocalDisk {
|
||||
Err(DiskError::VolumeExists)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
|
||||
let mut volumes = Vec::new();
|
||||
|
||||
@@ -7278,7 +7278,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(volumes)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
let meta = lstat(&volume_dir).await.map_err(to_volume_error)?;
|
||||
@@ -7294,7 +7294,7 @@ impl DiskAPI for LocalDisk {
|
||||
})
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> {
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
if !skip_access_checks(volume) {
|
||||
@@ -7318,7 +7318,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
|
||||
if !fi.metadata.is_empty() {
|
||||
let file_path = self.get_object_path(volume, path)?;
|
||||
@@ -7354,7 +7354,7 @@ impl DiskAPI for LocalDisk {
|
||||
Err(Error::other("Invalid Argument"))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::write_metadata");
|
||||
let p = self.get_object_path(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str())?;
|
||||
@@ -7381,7 +7381,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_version(
|
||||
&self,
|
||||
org_volume: &str,
|
||||
@@ -7484,7 +7484,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(fi)
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
|
||||
crate::hp_guard!("LocalDisk::read_xl");
|
||||
let file_path = self.get_object_path(volume, path)?;
|
||||
@@ -7495,7 +7495,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(RawFileInfo { buf })
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_version(
|
||||
&self,
|
||||
volume: &str,
|
||||
@@ -7723,7 +7723,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_versions(&self, volume: &str, versions: Vec<FileInfoVersions>, opts: DeleteOptions) -> Vec<Option<Error>> {
|
||||
let mut errs = Vec::with_capacity(versions.len());
|
||||
for _ in 0..versions.len() {
|
||||
@@ -7744,7 +7744,7 @@ impl DiskAPI for LocalDisk {
|
||||
errs
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
|
||||
let mut results = Vec::new();
|
||||
let mut found = 0;
|
||||
@@ -7810,7 +7810,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(results)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_volume(&self, volume: &str, force_delete: bool) -> Result<()> {
|
||||
let p = self.get_bucket_path(volume)?;
|
||||
|
||||
@@ -7843,7 +7843,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn disk_info(&self, _: &DiskInfoOptions) -> Result<DiskInfo> {
|
||||
let mut info = Cache::get(self.disk_info_cache.clone()).await?;
|
||||
info.nr_requests = self.nrrequests;
|
||||
@@ -7858,13 +7858,13 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
Ok(info)
|
||||
}
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
fn start_scan(&self) -> ScanGuard {
|
||||
self.scanning.fetch_add(1, Ordering::Release);
|
||||
ScanGuard(Arc::clone(&self.scanning))
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
|
||||
crate::hp_guard!("LocalDisk::read_metadata");
|
||||
let file_path = self.get_object_path(volume, path)?;
|
||||
@@ -7899,7 +7899,7 @@ async fn wait_for_startup_cleanup_signal(
|
||||
.is_ok()
|
||||
}
|
||||
|
||||
#[tracing::instrument]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInfo, bool)> {
|
||||
let drive_path = drive_path.to_string_lossy().to_string();
|
||||
check_path_length(&drive_path)?;
|
||||
|
||||
@@ -155,7 +155,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn make_volume(&self, volume: &str) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.make_volume(volume).await,
|
||||
@@ -163,7 +163,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.make_volumes(volumes).await,
|
||||
@@ -171,7 +171,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.list_volumes().await,
|
||||
@@ -186,7 +186,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_volume(&self, volume: &str, force_delete: bool) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.delete_volume(volume, force_delete).await,
|
||||
@@ -194,7 +194,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self, wr))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await,
|
||||
@@ -202,7 +202,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_version(
|
||||
&self,
|
||||
volume: &str,
|
||||
@@ -217,7 +217,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_versions(&self, volume: &str, versions: Vec<FileInfoVersions>, opts: DeleteOptions) -> Vec<Option<Error>> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.delete_versions(volume, versions, opts).await,
|
||||
@@ -225,7 +225,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.delete_paths(volume, paths).await,
|
||||
@@ -233,7 +233,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.write_metadata(_org_volume, volume, path, fi).await,
|
||||
@@ -241,7 +241,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.update_metadata(volume, path, fi, opts).await,
|
||||
@@ -249,7 +249,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_version(
|
||||
&self,
|
||||
_org_volume: &str,
|
||||
@@ -264,7 +264,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.read_xl(volume, path, read_data).await,
|
||||
@@ -272,7 +272,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self, fi))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_data(
|
||||
&self,
|
||||
src_volume: &str,
|
||||
@@ -287,7 +287,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn list_dir(&self, _origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.list_dir(_origvolume, volume, dir_path, count).await,
|
||||
@@ -295,7 +295,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.read_file(volume, path).await,
|
||||
@@ -303,7 +303,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.read_file_stream(volume, path, offset, length).await,
|
||||
@@ -311,7 +311,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.read_file_mmap_copy(volume, path, offset, length).await,
|
||||
@@ -337,7 +337,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.append_file(volume, path).await,
|
||||
@@ -345,7 +345,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: i64) -> Result<FileWriter> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.create_file(_origvolume, volume, path, _file_size).await,
|
||||
@@ -353,7 +353,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await,
|
||||
@@ -361,7 +361,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_parts(&self, bucket: &str, paths: &[String]) -> Result<Vec<ObjectPartInfo>> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.read_parts(bucket, paths).await,
|
||||
@@ -369,7 +369,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Bytes) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.rename_part(src_volume, src_path, dst_volume, dst_path, meta).await,
|
||||
@@ -381,7 +381,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.delete(volume, path, opt).await,
|
||||
@@ -389,7 +389,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.verify_file(volume, path, fi).await,
|
||||
@@ -397,7 +397,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.check_parts(volume, path, fi).await,
|
||||
@@ -405,7 +405,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.read_multiple(req).await,
|
||||
@@ -413,7 +413,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await,
|
||||
@@ -421,7 +421,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.read_all(volume, path).await,
|
||||
@@ -429,7 +429,7 @@ impl DiskAPI for Disk {
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo> {
|
||||
match self {
|
||||
Disk::Local(local_disk) => local_disk.disk_info(opts).await,
|
||||
|
||||
@@ -74,6 +74,7 @@ rustfs-notify = { workspace = true }
|
||||
rustfs-security-governance = { workspace = true }
|
||||
rustfs-storage-api = { workspace = true }
|
||||
rustfs-utils = { workspace = true, features = ["ip"] }
|
||||
rustix = { workspace = true, features = ["fs", "stdio"] }
|
||||
chrono = { workspace = true }
|
||||
flate2 = { workspace = true }
|
||||
glob = { workspace = true }
|
||||
|
||||
@@ -44,13 +44,12 @@ use rustfs_config::observability::{
|
||||
};
|
||||
use rustfs_config::{
|
||||
APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOG_FILENAME,
|
||||
DEFAULT_OBS_LOG_STDOUT_ENABLED, DEFAULT_OBS_LOGS_EXPORT_ENABLED, DEFAULT_OBS_METRICS_EXPORT_ENABLED,
|
||||
DEFAULT_OBS_PROFILING_EXPORT_ENABLED, DEFAULT_OBS_TRACES_EXPORT_ENABLED, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO,
|
||||
SERVICE_VERSION, USE_STDOUT,
|
||||
DEFAULT_OBS_LOGS_EXPORT_ENABLED, DEFAULT_OBS_METRICS_EXPORT_ENABLED, DEFAULT_OBS_PROFILING_EXPORT_ENABLED,
|
||||
DEFAULT_OBS_TRACES_EXPORT_ENABLED, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT,
|
||||
};
|
||||
use rustfs_utils::{
|
||||
get_env_bool, get_env_bool_with_aliases, get_env_f64, get_env_opt_str, get_env_opt_u64, get_env_str, get_env_u64,
|
||||
get_env_usize,
|
||||
get_env_bool, get_env_bool_with_aliases, get_env_f64, get_env_opt_bool, get_env_opt_str, get_env_opt_u64, get_env_str,
|
||||
get_env_u64, get_env_usize,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::env;
|
||||
@@ -310,7 +309,7 @@ impl OtelConfig {
|
||||
environment: Some(get_env_str(ENV_OBS_ENVIRONMENT, ENVIRONMENT)),
|
||||
// Local logging
|
||||
logger_level: Some(get_env_str(ENV_OBS_LOGGER_LEVEL, DEFAULT_LOG_LEVEL)),
|
||||
log_stdout_enabled: Some(get_env_bool(ENV_OBS_LOG_STDOUT_ENABLED, DEFAULT_OBS_LOG_STDOUT_ENABLED)),
|
||||
log_stdout_enabled: get_env_opt_bool(ENV_OBS_LOG_STDOUT_ENABLED),
|
||||
log_directory,
|
||||
log_filename: Some(get_env_nonempty_str(ENV_OBS_LOG_FILENAME, DEFAULT_OBS_LOG_FILENAME)),
|
||||
log_rotation_time,
|
||||
@@ -495,4 +494,19 @@ mod tests {
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stdout_mirror_environment_preserves_unset_true_and_false() {
|
||||
with_profiling_env_lock(|| {
|
||||
temp_env::with_var_unset(ENV_OBS_LOG_STDOUT_ENABLED, || {
|
||||
assert_eq!(OtelConfig::extract_otel_config_from_env(None).log_stdout_enabled, None);
|
||||
});
|
||||
temp_env::with_var(ENV_OBS_LOG_STDOUT_ENABLED, Some("true"), || {
|
||||
assert_eq!(OtelConfig::extract_otel_config_from_env(None).log_stdout_enabled, Some(true));
|
||||
});
|
||||
temp_env::with_var(ENV_OBS_LOG_STDOUT_ENABLED, Some("false"), || {
|
||||
assert_eq!(OtelConfig::extract_otel_config_from_env(None).log_stdout_enabled, Some(false));
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -68,6 +68,8 @@ pub enum TelemetryError {
|
||||
Io(String),
|
||||
#[error("Set permissions failed: {0}")]
|
||||
SetPermissions(String),
|
||||
#[error("Log sink conflict: {0}")]
|
||||
LogSinkConflict(String),
|
||||
}
|
||||
|
||||
impl From<std::io::Error> for TelemetryError {
|
||||
|
||||
@@ -47,7 +47,7 @@ use rustfs_config::observability::{
|
||||
DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP,
|
||||
DEFAULT_OBS_LOG_ZSTD_WORKERS,
|
||||
};
|
||||
use rustfs_config::{APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOG_STDOUT_ENABLED};
|
||||
use rustfs_config::{APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME};
|
||||
use serde_json::Value as JsonValue;
|
||||
use std::sync::Arc;
|
||||
use std::{collections::BTreeMap, fmt};
|
||||
@@ -74,6 +74,59 @@ const STDERR_WARNING_PREFIX: &str = "[WARN]";
|
||||
const REQUEST_ID_CANONICAL: &str = "request_id";
|
||||
const REQUEST_ID_COMPAT: &str = "request-id";
|
||||
|
||||
pub(super) fn resolve_file_stdout_mirror(configured: Option<bool>, is_production: bool) -> bool {
|
||||
configured.unwrap_or(!is_production)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(super) fn validate_stdout_sink(file_appender: &RollingAppender) -> Result<(), TelemetryError> {
|
||||
use std::os::fd::AsFd;
|
||||
|
||||
let stdout_stat = rustix::fs::fstat(std::io::stdout().as_fd())
|
||||
.map_err(|error| TelemetryError::LogSinkConflict(format!("failed to inspect stdout file descriptor: {error}")))?;
|
||||
if !rustix::fs::FileType::from_raw_mode(stdout_stat.st_mode).is_file() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let stdout_device = {
|
||||
#[cfg(any(target_os = "linux", target_os = "android"))]
|
||||
{
|
||||
stdout_stat.st_dev
|
||||
}
|
||||
#[cfg(not(any(target_os = "linux", target_os = "android")))]
|
||||
{
|
||||
u64::try_from(stdout_stat.st_dev).map_err(|error| {
|
||||
TelemetryError::LogSinkConflict(format!("stdout file descriptor returned an invalid device identifier: {error}"))
|
||||
})?
|
||||
}
|
||||
};
|
||||
let active_identity = file_appender
|
||||
.active_file_identity()
|
||||
.map_err(|error| TelemetryError::Io(error.to_string()))?;
|
||||
validate_distinct_file_identities((stdout_device, stdout_stat.st_ino), active_identity, &file_appender.active_file_path())
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(super) fn validate_stdout_sink(_file_appender: &RollingAppender) -> Result<(), TelemetryError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn validate_distinct_file_identities(
|
||||
stdout_identity: (u64, u64),
|
||||
active_identity: (u64, u64),
|
||||
active_log_path: &std::path::Path,
|
||||
) -> Result<(), TelemetryError> {
|
||||
if stdout_identity == active_identity {
|
||||
return Err(TelemetryError::LogSinkConflict(format!(
|
||||
"stdout and rolling log resolve to the same file {}; route stdout to journald or choose a different log file",
|
||||
active_log_path.display()
|
||||
)));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn resolve_log_cleanup_interval_seconds(config: &OtelConfig) -> u64 {
|
||||
match config.log_cleanup_interval_seconds {
|
||||
Some(0) => {
|
||||
@@ -353,6 +406,7 @@ fn init_file_logging_internal(
|
||||
|
||||
let file_appender =
|
||||
RollingAppender::new(log_directory, log_filename.to_string(), rotation, max_single_file_size, match_mode)?;
|
||||
validate_stdout_sink(&file_appender)?;
|
||||
|
||||
let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
|
||||
|
||||
@@ -367,7 +421,7 @@ fn init_file_logging_internal(
|
||||
// Optional stdout mirror: enabled explicitly via `log_stdout_enabled`, or
|
||||
// unconditionally in non-production environments so developers still see
|
||||
// immediate terminal output while file rotation remains enabled.
|
||||
let (stdout_layer, stdout_guard) = if config.log_stdout_enabled.unwrap_or(DEFAULT_OBS_LOG_STDOUT_ENABLED) || !is_production {
|
||||
let (stdout_layer, stdout_guard) = if resolve_file_stdout_mirror(config.log_stdout_enabled, is_production) {
|
||||
let (stdout_nb, stdout_guard) = tracing_appender::non_blocking(std::io::stdout());
|
||||
let enable_color = std::io::stdout().is_terminal();
|
||||
(Some(build_json_log_layer(stdout_nb, enable_color, span_events)), Some(stdout_guard))
|
||||
@@ -802,6 +856,127 @@ mod tests {
|
||||
)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stdout_mirror_configuration_overrides_environment_default() {
|
||||
assert!(!resolve_file_stdout_mirror(None, true));
|
||||
assert!(resolve_file_stdout_mirror(None, false));
|
||||
assert!(resolve_file_stdout_mirror(Some(true), true));
|
||||
assert!(resolve_file_stdout_mirror(Some(true), false));
|
||||
assert!(!resolve_file_stdout_mirror(Some(false), true));
|
||||
assert!(!resolve_file_stdout_mirror(Some(false), false));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn stdout_sink_validation_rejects_file_aliases() {
|
||||
use std::fs::{self, File};
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
let temp_dir = tempdir().expect("create temp dir");
|
||||
let active_path = temp_dir.path().join("rustfs.log");
|
||||
let hardlink_path = temp_dir.path().join("stdout-hardlink.log");
|
||||
let symlink_path = temp_dir.path().join("stdout-symlink.log");
|
||||
File::create(&active_path).expect("create active log");
|
||||
fs::hard_link(&active_path, &hardlink_path).expect("create hardlink");
|
||||
std::os::unix::fs::symlink(&active_path, &symlink_path).expect("create symlink");
|
||||
|
||||
let active_metadata = fs::metadata(&active_path).expect("stat active log");
|
||||
for alias in [&active_path, &hardlink_path, &symlink_path] {
|
||||
let alias_metadata = fs::metadata(alias).expect("stat stdout alias");
|
||||
assert!(matches!(
|
||||
validate_distinct_file_identities(
|
||||
(alias_metadata.dev(), alias_metadata.ino()),
|
||||
(active_metadata.dev(), active_metadata.ino()),
|
||||
&active_path,
|
||||
),
|
||||
Err(TelemetryError::LogSinkConflict(_))
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn stdout_sink_validation_allows_distinct_files() {
|
||||
use std::fs::{self, File};
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
let temp_dir = tempdir().expect("create temp dir");
|
||||
let stdout_path = temp_dir.path().join("stdout.log");
|
||||
let active_path = temp_dir.path().join("rustfs.log");
|
||||
File::create(&stdout_path).expect("create stdout log");
|
||||
File::create(&active_path).expect("create active log");
|
||||
|
||||
let stdout_metadata = fs::metadata(&stdout_path).expect("stat stdout log");
|
||||
let active_metadata = fs::metadata(&active_path).expect("stat active log");
|
||||
assert!(
|
||||
validate_distinct_file_identities(
|
||||
(stdout_metadata.dev(), stdout_metadata.ino()),
|
||||
(active_metadata.dev(), active_metadata.ino()),
|
||||
&active_path,
|
||||
)
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn file_logging_initialization_rejects_stdout_alias() {
|
||||
const CHILD_ENV: &str = "RUSTFS_TEST_STDOUT_LOG_ALIAS_CHILD";
|
||||
const PATH_ENV: &str = "RUSTFS_TEST_STDOUT_LOG_ALIAS_PATH";
|
||||
|
||||
if std::env::var_os(CHILD_ENV).is_some() {
|
||||
use std::fs::OpenOptions;
|
||||
|
||||
let active_path = std::path::PathBuf::from(std::env::var_os(PATH_ENV).expect("child log path must be set"));
|
||||
let stdout_file = OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(&active_path)
|
||||
.expect("open child stdout log");
|
||||
rustix::stdio::dup2_stdout(&stdout_file).expect("redirect child stdout");
|
||||
|
||||
let config = OtelConfig {
|
||||
log_filename: active_path.file_name().map(|name| name.to_string_lossy().into_owned()),
|
||||
log_stdout_enabled: Some(false),
|
||||
..OtelConfig::default()
|
||||
};
|
||||
let log_directory = active_path.parent().expect("active log must have parent");
|
||||
let runtime = tokio::runtime::Runtime::new().expect("create Tokio runtime");
|
||||
runtime.block_on(async {
|
||||
assert!(matches!(
|
||||
init_file_logging_internal(
|
||||
&config,
|
||||
log_directory.to_str().expect("log directory must be UTF-8"),
|
||||
"info",
|
||||
true,
|
||||
),
|
||||
Err(TelemetryError::LogSinkConflict(_))
|
||||
));
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
let temp_dir = tempdir().expect("create parent temp dir");
|
||||
let active_path = temp_dir.path().join("rustfs.log");
|
||||
let output = std::process::Command::new(std::env::current_exe().expect("resolve current test executable"))
|
||||
.args([
|
||||
"--exact",
|
||||
"telemetry::local::tests::file_logging_initialization_rejects_stdout_alias",
|
||||
"--nocapture",
|
||||
])
|
||||
.env(CHILD_ENV, "1")
|
||||
.env(PATH_ENV, &active_path)
|
||||
.output()
|
||||
.expect("run isolated stdout alias child test");
|
||||
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"stdout alias child failed: stdout={} stderr={}",
|
||||
String::from_utf8_lossy(&output.stdout),
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_file_logging_fallback_warning_message_is_actionable() {
|
||||
let message = format_file_logging_fallback_warning(
|
||||
|
||||
@@ -60,7 +60,7 @@ use opentelemetry_sdk::{
|
||||
use percent_encoding::percent_decode_str;
|
||||
use rustfs_config::observability::{DEFAULT_OBS_LOG_MATCH_MODE, DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES};
|
||||
use rustfs_config::{
|
||||
APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOG_STDOUT_ENABLED, DEFAULT_OBS_LOGS_EXPORT_ENABLED,
|
||||
APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOGS_EXPORT_ENABLED,
|
||||
DEFAULT_OBS_METRICS_EXPORT_ENABLED, DEFAULT_OBS_TRACES_EXPORT_ENABLED, METER_INTERVAL, SAMPLE_RATIO,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
@@ -200,6 +200,7 @@ pub(super) fn init_observability_http(
|
||||
|
||||
let file_appender =
|
||||
RollingAppender::new(log_directory, log_filename.to_string(), rotation, max_single_file_size, match_mode)?;
|
||||
crate::telemetry::local::validate_stdout_sink(&file_appender)?;
|
||||
|
||||
let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
|
||||
let file_layer = build_json_log_layer(non_blocking, false, span_events.clone());
|
||||
@@ -220,8 +221,10 @@ pub(super) fn init_observability_http(
|
||||
log_directory,
|
||||
rotation = %rotation_str,
|
||||
keep_files,
|
||||
stdout_mirror_enabled = config.log_stdout_enabled.unwrap_or(DEFAULT_OBS_LOG_STDOUT_ENABLED)
|
||||
|| !is_production,
|
||||
stdout_mirror_enabled = crate::telemetry::local::resolve_file_stdout_mirror(
|
||||
config.log_stdout_enabled,
|
||||
is_production,
|
||||
),
|
||||
logger_level,
|
||||
is_production,
|
||||
"Initialized local logging fallback for observability"
|
||||
@@ -243,7 +246,7 @@ pub(super) fn init_observability_http(
|
||||
|
||||
// Optional stdout mirror (matching init_file_logging_internal logic)
|
||||
// This is separate from OTLP stdout logic. If file logging is enabled, we honor its stdout rules.
|
||||
if force_stdout_logging || config.log_stdout_enabled.unwrap_or(DEFAULT_OBS_LOG_STDOUT_ENABLED) || !is_production {
|
||||
if force_stdout_logging || crate::telemetry::local::resolve_file_stdout_mirror(config.log_stdout_enabled, is_production) {
|
||||
let (stdout_nb, stdout_g) = tracing_appender::non_blocking(std::io::stdout());
|
||||
stdout_guard = Some(stdout_g);
|
||||
stdout_layer_opt = Some(build_json_log_layer(stdout_nb, std::io::stdout().is_terminal(), span_events));
|
||||
|
||||
@@ -133,10 +133,22 @@ impl RollingAppender {
|
||||
Ok(appender)
|
||||
}
|
||||
|
||||
fn active_file_path(&self) -> PathBuf {
|
||||
pub(crate) fn active_file_path(&self) -> PathBuf {
|
||||
self.dir.join(&self.filename)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn active_file_identity(&self) -> io::Result<(u64, u64)> {
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
let metadata = self
|
||||
.file
|
||||
.as_ref()
|
||||
.ok_or_else(|| io::Error::other("rolling log file is not open"))?
|
||||
.metadata()?;
|
||||
Ok((metadata.dev(), metadata.ino()))
|
||||
}
|
||||
|
||||
fn open_file(&mut self) -> io::Result<()> {
|
||||
if self.file.is_some() {
|
||||
return Ok(());
|
||||
|
||||
Reference in New Issue
Block a user