gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 05:06:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(obs): unify metrics initialization and fix exporter move error (#851)

Browse Source 
* feat(obs): unify metrics initialization and fix exporter move error

- Fix Rust E0382 (use after move) by removing duplicate MetricExporter consumption.
- Consolidate MeterProvider construction into single Recorder builder path.
- Remove redundant Recorder::builder(...).install_global() call.
- Ensure PeriodicReader setup is performed only once (HTTP + optional stdout).
- Set global meter provider and metrics recorder exactly once.
- Preserve existing behavior for stdout/file vs HTTP modes.
- Minor cleanup: consistent resource reuse and interval handling.

* update telemetry.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* fix

* fix

* fix

* fix: modify logger level from error to event

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
houseme
2025-11-14 00:50:07 +08:00
committed by GitHub
parent 93090adf7c
commit 51584986e1
12 changed files with 502 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/obs/src/telemetry.rs
+70 -97
View File
@@ -13,45 +13,44 @@
// limitations under the License.
use crate::config::OtelConfig;
use crate::global::IS_OBSERVABILITY_ENABLED;
use crate::global::OBSERVABILITY_METRIC_ENABLED;
use crate::{Recorder, TelemetryError};
use flexi_logger::{DeferredNow, Record, WriteMode, WriteMode::AsyncWith, style};
use metrics::counter;
use nu_ansi_term::Color;
use opentelemetry::trace::TracerProvider;
use opentelemetry::{KeyValue, global};
use opentelemetry::{KeyValue, global, trace::TracerProvider};
use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
use opentelemetry_otlp::{Compression, Protocol, WithExportConfig, WithHttpConfig};
use opentelemetry_sdk::{
Resource,
logs::SdkLoggerProvider,
metrics::{MeterProviderBuilder, PeriodicReader, SdkMeterProvider},
metrics::{PeriodicReader, SdkMeterProvider},
trace::{RandomIdGenerator, Sampler, SdkTracerProvider},
};
use opentelemetry_semantic_conventions::{
SCHEMA_URL,
attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_VERSION as OTEL_SERVICE_VERSION},
};
use rustfs_config::observability::{ENV_OBS_LOG_FLUSH_MS, ENV_OBS_LOG_MESSAGE_CAPA, ENV_OBS_LOG_POOL_CAPA};
use rustfs_config::{
APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, DEFAULT_OBS_LOG_STDOUT_ENABLED, ENVIRONMENT, METER_INTERVAL,
SAMPLE_RATIO, SERVICE_VERSION,
observability::{
DEFAULT_OBS_ENVIRONMENT_PRODUCTION, DEFAULT_OBS_LOG_FLUSH_MS, DEFAULT_OBS_LOG_MESSAGE_CAPA, DEFAULT_OBS_LOG_POOL_CAPA,
ENV_OBS_LOG_DIRECTORY,
ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_FLUSH_MS, ENV_OBS_LOG_MESSAGE_CAPA, ENV_OBS_LOG_POOL_CAPA,
},
};
use rustfs_utils::{get_env_u64, get_env_usize, get_local_ip_with_default};
use smallvec::SmallVec;
use std::borrow::Cow;
use std::io::IsTerminal;
use std::time::Duration;
use std::{env, fs};
use std::{borrow::Cow, env, fs, io::IsTerminal, time::Duration};
use tracing::info;
use tracing_error::ErrorLayer;
use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
use tracing_subscriber::fmt::format::FmtSpan;
use tracing_subscriber::fmt::time::LocalTime;
use tracing_subscriber::{EnvFilter, Layer, layer::SubscriberExt, util::SubscriberInitExt};
use tracing_subscriber::{
EnvFilter, Layer,
fmt::{format::FmtSpan, time::LocalTime},
layer::SubscriberExt,
util::SubscriberInitExt,
};
/// A guard object that manages the lifecycle of OpenTelemetry components.
///
@@ -69,9 +68,7 @@ pub struct OtelGuard {
tracer_provider: Option<SdkTracerProvider>,
meter_provider: Option<SdkMeterProvider>,
logger_provider: Option<SdkLoggerProvider>,
// Add a flexi_logger handle to keep the logging alive
flexi_logger_handles: Option<flexi_logger::LoggerHandle>,
// WorkerGuard for writing tracing files
tracing_guard: Option<tracing_appender::non_blocking::WorkerGuard>,
}
@@ -112,35 +109,12 @@ impl Drop for OtelGuard {
}
if let Some(guard) = self.tracing_guard.take() {
// The guard will be dropped here, flushing any remaining logs
drop(guard);
println!("Tracing guard dropped, flushing logs.");
}
}
}
#[derive(Debug)]
pub enum TelemetryError {
BuildSpanExporter(String),
BuildMetricExporter(String),
BuildLogExporter(String),
InstallMetricsRecorder(String),
SubscriberInit(String),
}
impl std::fmt::Display for TelemetryError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
TelemetryError::BuildSpanExporter(e) => write!(f, "Span exporter build failed: {e}"),
TelemetryError::BuildMetricExporter(e) => write!(f, "Metric exporter build failed: {e}"),
TelemetryError::BuildLogExporter(e) => write!(f, "Log exporter build failed: {e}"),
TelemetryError::InstallMetricsRecorder(e) => write!(f, "Install metrics recorder failed: {e}"),
TelemetryError::SubscriberInit(e) => write!(f, "Tracing subscriber init failed: {e}"),
}
}
}
impl std::error::Error for TelemetryError {}
/// create OpenTelemetry Resource
fn resource(config: &OtelConfig) -> Resource {
Resource::builder()
@@ -170,16 +144,16 @@ fn create_periodic_reader(interval: u64) -> PeriodicReader<opentelemetry_stdout:
}
// Read the AsyncWith parameter from the environment variable
fn get_env_async_with() -> Option<WriteMode> {
fn get_env_async_with() -> WriteMode {
let pool_capa = get_env_usize(ENV_OBS_LOG_POOL_CAPA, DEFAULT_OBS_LOG_POOL_CAPA);
let message_capa = get_env_usize(ENV_OBS_LOG_MESSAGE_CAPA, DEFAULT_OBS_LOG_MESSAGE_CAPA);
let flush_ms = get_env_u64(ENV_OBS_LOG_FLUSH_MS, DEFAULT_OBS_LOG_FLUSH_MS);
Some(AsyncWith {
AsyncWith {
pool_capa,
message_capa,
flush_interval: Duration::from_millis(flush_ms),
})
}
}
fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter {
@@ -200,12 +174,9 @@ fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilte
fn format_with_color(w: &mut dyn std::io::Write, now: &mut DeferredNow, record: &Record) -> Result<(), std::io::Error> {
let level = record.level();
let level_style = style(level);
// Get the current thread information
let binding = std::thread::current();
let thread_name = binding.name().unwrap_or("unnamed");
let thread_id = format!("{:?}", std::thread::current().id());
writeln!(
w,
"[{}] {} [{}] [{}:{}] [{}:{}] {}",
@@ -224,12 +195,9 @@ fn format_with_color(w: &mut dyn std::io::Write, now: &mut DeferredNow, record:
#[inline(never)]
fn format_for_file(w: &mut dyn std::io::Write, now: &mut DeferredNow, record: &Record) -> Result<(), std::io::Error> {
let level = record.level();
// Get the current thread information
let binding = std::thread::current();
let thread_name = binding.name().unwrap_or("unnamed");
let thread_id = format!("{:?}", std::thread::current().id());
writeln!(
w,
"[{}] {} [{}] [{}:{}] [{}:{}] {}",
@@ -247,9 +215,7 @@ fn format_for_file(w: &mut dyn std::io::Write, now: &mut DeferredNow, record: &R
/// stdout + span information (fix: retain WorkerGuard to avoid releasing after initialization)
fn init_stdout_logging(_config: &OtelConfig, logger_level: &str, is_production: bool) -> OtelGuard {
let env_filter = build_env_filter(logger_level, None);
let (nb, guard) = tracing_appender::non_blocking(std::io::stdout());
let enable_color = std::io::stdout().is_terminal();
let fmt_layer = tracing_subscriber::fmt::layer()
.with_timer(LocalTime::rfc_3339())
@@ -264,17 +230,15 @@ fn init_stdout_logging(_config: &OtelConfig, logger_level: &str, is_production:
.with_current_span(true)
.with_span_list(true)
.with_span_events(if is_production { FmtSpan::CLOSE } else { FmtSpan::FULL });
tracing_subscriber::registry()
.with(env_filter)
.with(ErrorLayer::default())
.with(fmt_layer)
.init();
IS_OBSERVABILITY_ENABLED.set(false).ok();
OBSERVABILITY_METRIC_ENABLED.set(false).ok();
counter!("rustfs.start.total").increment(1);
info!("Init stdout logging (level: {})", logger_level);
OtelGuard {
tracer_provider: None,
meter_provider: None,
@@ -285,27 +249,49 @@ fn init_stdout_logging(_config: &OtelConfig, logger_level: &str, is_production:
}
/// File rolling log (size switching + number retained)
fn init_file_logging(config: &OtelConfig, logger_level: &str, is_production: bool) -> OtelGuard {
use flexi_logger::{
Age, Cleanup, Criterion, FileSpec, LogSpecification, Naming,
WriteMode::{AsyncWith, BufferAndFlush},
};
fn init_file_logging(config: &OtelConfig, logger_level: &str, is_production: bool) -> Result<OtelGuard, TelemetryError> {
use flexi_logger::{Age, Cleanup, Criterion, FileSpec, LogSpecification, Naming};
let service_name = config.service_name.as_deref().unwrap_or(APP_NAME);
let default_log_directory = rustfs_utils::dirs::get_log_directory_to_string(ENV_OBS_LOG_DIRECTORY);
let log_directory = config.log_directory.as_deref().unwrap_or(default_log_directory.as_str());
let log_filename = config.log_filename.as_deref().unwrap_or(service_name);
let keep_files = config.log_keep_files.unwrap_or(DEFAULT_LOG_KEEP_FILES);
if let Err(e) = fs::create_dir_all(log_directory) {
eprintln!("ERROR: create log dir '{}': {e}", log_directory);
return Err(TelemetryError::Io(e.to_string()));
}
#[cfg(unix)]
{
use std::fs::Permissions;
use std::os::unix::fs::PermissionsExt;
let _ = fs::set_permissions(log_directory, Permissions::from_mode(0o755));
let desired: u32 = 0o755;
match fs::metadata(log_directory) {
Ok(meta) => {
let current = meta.permissions().mode() & 0o777;
// Only tighten to 0755 if existing permissions are looser than target, avoid loosening
if (current & !desired) != 0 {
if let Err(e) = fs::set_permissions(log_directory, Permissions::from_mode(desired)) {
return Err(TelemetryError::SetPermissions(format!(
"dir='{}', want={:#o}, have={:#o}, err={}",
log_directory, desired, current, e
)));
}
// Second verification
if let Ok(meta2) = fs::metadata(log_directory) {
let after = meta2.permissions().mode() & 0o777;
if after != desired {
return Err(TelemetryError::SetPermissions(format!(
"dir='{}', want={:#o}, after={:#o}",
log_directory, desired, after
)));
}
}
}
}
Err(e) => {
return Err(TelemetryError::Io(format!("stat '{}' failed: {}", log_directory, e)));
}
}
}
// parsing level
@@ -345,25 +331,10 @@ fn init_file_logging(config: &OtelConfig, logger_level: &str, is_production: boo
};
// write mode
let write_mode = get_env_async_with().unwrap_or(if is_production {
AsyncWith {
pool_capa: DEFAULT_OBS_LOG_POOL_CAPA,
message_capa: DEFAULT_OBS_LOG_MESSAGE_CAPA,
flush_interval: Duration::from_millis(DEFAULT_OBS_LOG_FLUSH_MS),
}
} else {
BufferAndFlush
});
let write_mode = get_env_async_with();
// Build
let mut builder = flexi_logger::Logger::try_with_env_or_str(logger_level)
.unwrap_or_else(|e| {
if !is_production {
eprintln!("WARNING: Invalid logger configuration '{logger_level}': {e:?}");
eprintln!("Falling back to default configuration with level: {DEFAULT_LOG_LEVEL}");
}
flexi_logger::Logger::with(log_spec.clone())
})
.unwrap_or(flexi_logger::Logger::with(log_spec.clone()))
.format_for_stderr(format_with_color)
.format_for_stdout(format_with_color)
.format_for_files(format_for_file)
@@ -379,7 +350,7 @@ fn init_file_logging(config: &OtelConfig, logger_level: &str, is_production: boo
.use_utc();
// Optional copy to stdout (for local observation)
if config.log_stdout_enabled.unwrap_or(DEFAULT_OBS_LOG_STDOUT_ENABLED) {
if config.log_stdout_enabled.unwrap_or(DEFAULT_OBS_LOG_STDOUT_ENABLED) || !is_production {
builder = builder.duplicate_to_stdout(flexi_logger::Duplicate::All);
} else {
builder = builder.duplicate_to_stdout(flexi_logger::Duplicate::None);
@@ -393,20 +364,20 @@ fn init_file_logging(config: &OtelConfig, logger_level: &str, is_production: boo
}
};
IS_OBSERVABILITY_ENABLED.set(false).ok();
OBSERVABILITY_METRIC_ENABLED.set(false).ok();
counter!("rustfs.start.total").increment(1);
info!(
"Init file logging at '{}', roll size {:?}MB, keep {}",
log_directory, config.log_rotation_size_mb, keep_files
);
OtelGuard {
Ok(OtelGuard {
tracer_provider: None,
meter_provider: None,
logger_provider: None,
flexi_logger_handles: handle,
tracing_guard: None,
}
})
}
/// Observability (HTTP export, supports three sub-endpoints; if not, fallback to unified endpoint)
@@ -464,24 +435,26 @@ fn init_observability_http(config: &OtelConfig, logger_level: &str, is_productio
.build()
.map_err(|e| TelemetryError::BuildMetricExporter(e.to_string()))?;
let meter_interval = config.meter_interval.unwrap_or(METER_INTERVAL);
let mut builder = MeterProviderBuilder::default().with_resource(res.clone());
builder = builder.with_reader(
PeriodicReader::builder(exporter)
.with_interval(Duration::from_secs(meter_interval))
.build(),
);
if use_stdout {
builder = builder.with_reader(create_periodic_reader(meter_interval));
}
let provider = builder.build();
let (provider, recorder) = Recorder::builder(service_name.clone())
.with_meter_provider(|b| {
let b = b.with_resource(res.clone()).with_reader(
PeriodicReader::builder(exporter)
.with_interval(Duration::from_secs(meter_interval))
.build(),
);
if use_stdout {
b.with_reader(create_periodic_reader(meter_interval))
} else {
b
}
})
.build();
global::set_meter_provider(provider.clone());
metrics::set_global_recorder(recorder).map_err(|e| TelemetryError::InstallMetricsRecorder(e.to_string()))?;
provider
};
// metrics crate -> OTel
let _ = metrics_exporter_opentelemetry::Recorder::builder(service_name.clone()).install_global();
// Logger（HTTP）
let logger_provider = {
let exporter = opentelemetry_otlp::LogExporter::builder()
@@ -536,7 +509,7 @@ fn init_observability_http(config: &OtelConfig, logger_level: &str, is_productio
.with(MetricsLayer::new(meter_provider.clone()))
.init();
IS_OBSERVABILITY_ENABLED.set(true).ok();
OBSERVABILITY_METRIC_ENABLED.set(true).ok();
counter!("rustfs.start.total").increment(1);
info!(
"Init observability (HTTP): trace='{}', metric='{}', log='{}'",
@@ -571,7 +544,7 @@ pub(crate) fn init_telemetry(config: &OtelConfig) -> Result<OtelGuard, Telemetry
// Rule 2: The user has explicitly customized the log directory (determined by whether ENV_OBS_LOG_DIRECTORY is set)
let user_set_log_dir = env::var(ENV_OBS_LOG_DIRECTORY).is_ok();
if user_set_log_dir {
return Ok(init_file_logging(config, logger_level, is_production));
return init_file_logging(config, logger_level, is_production);
}
// Rule 1: Default stdout (error level)