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

improve code for obs

Browse Source
This commit is contained in:
houseme
2025-05-29 21:10:04 +08:00
parent ace04bcd97
commit 3ceec7b5f2
9 changed files with 755 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/config/src/constants/app.rs
+42 -13
View File
@@ -56,14 +56,13 @@ pub const DEFAULT_ACCESS_KEY: &str = "rustfsadmin";
/// Example: RUSTFS_SECRET_KEY=rustfsadmin
/// Example: --secret-key rustfsadmin
pub const DEFAULT_SECRET_KEY: &str = "rustfsadmin";
/// Default configuration file for observability
/// Default value: config/obs.toml
/// Environment variable: RUSTFS_OBS_CONFIG
/// Command line argument: --obs-config
/// Example: RUSTFS_OBS_CONFIG=config/obs.toml
/// Example: --obs-config config/obs.toml
/// Example: --obs-config /etc/rustfs/obs.toml
pub const DEFAULT_OBS_CONFIG: &str = "./deploy/config/obs.toml";
/// Default OBS configuration endpoint
/// Environment variable: DEFAULT_OBS_ENDPOINT
/// Command line argument: --obs-endpoint
/// Example: DEFAULT_OBS_ENDPOINT="http://localost:4317"
/// Example: --obs-endpoint http://localost:4317
pub const DEFAULT_OBS_ENDPOINT: &str = "";
/// Default TLS key for rustfs
/// This is the default key for TLS.
@@ -90,6 +89,41 @@ pub const DEFAULT_CONSOLE_PORT: u16 = 9001;
/// This is the default address for rustfs console.
pub const DEFAULT_CONSOLE_ADDRESS: &str = concat!(":", DEFAULT_CONSOLE_PORT);
/// Default log filename for rustfs
/// This is the default log filename for rustfs.
/// It is used to store the logs of the application.
/// Default value: rustfs.log
/// Environment variable: RUSTFS_OBSERVABILITY_LOG_FILENAME
pub const DEFAULT_LOG_FILENAME: &str = "rustfs.log";
/// Default log directory for rustfs
/// This is the default log directory for rustfs.
/// It is used to store the logs of the application.
/// Default value: logs
/// Environment variable: RUSTFS_OBSERVABILITY_LOG_DIRECTORY
pub const DEFAULT_LOG_DIR: &str = "logs";
/// Default log rotation size mb for rustfs
/// This is the default log rotation size for rustfs.
/// It is used to rotate the logs of the application.
/// Default value: 100 MB
/// Environment variable: RUSTFS_OBSERVABILITY_LOG_ROTATION_SIZE_MB
pub const DEFAULT_LOG_ROTATION_SIZE_MB: u64 = 100;
/// Default log rotation time for rustfs
/// This is the default log rotation time for rustfs.
/// It is used to rotate the logs of the application.
/// Default value: hour, eg: day,hour,minute,second
/// Environment variable: RUSTFS_OBSERVABILITY_LOG_ROTATION_TIME
pub const DEFAULT_LOG_ROTATION_TIME: &str = "day";
/// Default log keep files for rustfs
/// This is the default log keep files for rustfs.
/// It is used to keep the logs of the application.
/// Default value: 30
/// Environment variable: RUSTFS_OBSERVABILITY_LOG_KEEP_FILES
pub const DEFAULT_LOG_KEEP_FILES: u16 = 30;
#[cfg(test)]
mod tests {
use super::*;
@@ -154,10 +188,6 @@ mod tests {
#[test]
fn test_file_path_constants() {
// Test file path related constants
assert_eq!(DEFAULT_OBS_CONFIG, "./deploy/config/obs.toml");
assert!(DEFAULT_OBS_CONFIG.ends_with(".toml"), "Config file should be TOML format");
assert_eq!(RUSTFS_TLS_KEY, "rustfs_key.pem");
assert!(RUSTFS_TLS_KEY.ends_with(".pem"), "TLS key should be PEM format");
@@ -219,7 +249,6 @@ mod tests {
ENVIRONMENT,
DEFAULT_ACCESS_KEY,
DEFAULT_SECRET_KEY,
DEFAULT_OBS_CONFIG,
RUSTFS_TLS_KEY,
RUSTFS_TLS_CERT,
DEFAULT_ADDRESS,
crates/obs/Cargo.toml
+2 -2
View File
@@ -20,8 +20,8 @@ kafka = ["dep:rdkafka"]
rustfs-config = { workspace = true }
async-trait = { workspace = true }
chrono = { workspace = true }
config = { workspace = true }
local-ip-address = { workspace = true }
flexi_logger = { workspace = true, features = ["trc", "kv"] }
nu-ansi-term = { workspace = true }
nvml-wrapper = { workspace = true, optional = true }
opentelemetry = { workspace = true }
opentelemetry-appender-tracing = { workspace = true, features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] }
crates/obs/src/config.rs
+112 -111
View File
@@ -1,5 +1,7 @@
use config::{Config, File, FileFormat};
use rustfs_config::{APP_NAME, DEFAULT_LOG_LEVEL, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT};
use rustfs_config::{
APP_NAME, DEFAULT_LOG_DIR, DEFAULT_LOG_FILENAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, DEFAULT_LOG_ROTATION_SIZE_MB,
DEFAULT_LOG_ROTATION_TIME, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT,
};
use serde::{Deserialize, Serialize};
use std::env;
@@ -22,60 +24,94 @@ pub struct OtelConfig {
pub environment: Option<String>, // Environment
pub logger_level: Option<String>, // Logger level
pub local_logging_enabled: Option<bool>, // Local logging enabled
}
/// Helper function: Extract observable configuration from environment variables
fn extract_otel_config_from_env() -> OtelConfig {
let endpoint = env::var("RUSTFS_OBSERVABILITY_ENDPOINT").unwrap_or_else(|_| "".to_string());
let mut use_stdout = env::var("RUSTFS_OBSERVABILITY_USE_STDOUT")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(USE_STDOUT));
if endpoint.is_empty() {
use_stdout = Some(true);
}
OtelConfig {
endpoint,
use_stdout,
sample_ratio: env::var("RUSTFS_OBSERVABILITY_SAMPLE_RATIO")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(SAMPLE_RATIO)),
meter_interval: env::var("RUSTFS_OBSERVABILITY_METER_INTERVAL")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(METER_INTERVAL)),
service_name: env::var("RUSTFS_OBSERVABILITY_SERVICE_NAME")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(APP_NAME.to_string())),
service_version: env::var("RUSTFS_OBSERVABILITY_SERVICE_VERSION")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(SERVICE_VERSION.to_string())),
environment: env::var("RUSTFS_OBSERVABILITY_ENVIRONMENT")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(ENVIRONMENT.to_string())),
logger_level: env::var("RUSTFS_OBSERVABILITY_LOGGER_LEVEL")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(DEFAULT_LOG_LEVEL.to_string())),
local_logging_enabled: env::var("RUSTFS_OBSERVABILITY_LOCAL_LOGGING_ENABLED")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(false)),
}
// 新增 flexi_logger 相关配置
pub log_directory: Option<String>, // 日志文件目录
pub log_filename: Option<String>, // 日志文件名称
pub log_rotation_size_mb: Option<u64>, // 日志文件大小切割阈值 (MB)
pub log_rotation_time: Option<String>, // 日志按时间切割 (Hour, Day)
pub log_keep_files: Option<u16>, // 保留日志文件数量
}
impl OtelConfig {
/// Helper function: Extract observable configuration from environment variables
pub fn extract_otel_config_from_env(endpoint: Option<String>) -> OtelConfig {
let endpoint = if let Some(endpoint) = endpoint {
if endpoint.is_empty() {
env::var("RUSTFS_OBSERVABILITY_ENDPOINT").unwrap_or_else(|_| "".to_string())
} else {
endpoint
}
} else {
env::var("RUSTFS_OBSERVABILITY_ENDPOINT").unwrap_or_else(|_| "".to_string())
};
let mut use_stdout = env::var("RUSTFS_OBSERVABILITY_USE_STDOUT")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(USE_STDOUT));
if endpoint.is_empty() {
use_stdout = Some(true);
}
OtelConfig {
endpoint,
use_stdout,
sample_ratio: env::var("RUSTFS_OBSERVABILITY_SAMPLE_RATIO")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(SAMPLE_RATIO)),
meter_interval: env::var("RUSTFS_OBSERVABILITY_METER_INTERVAL")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(METER_INTERVAL)),
service_name: env::var("RUSTFS_OBSERVABILITY_SERVICE_NAME")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(APP_NAME.to_string())),
service_version: env::var("RUSTFS_OBSERVABILITY_SERVICE_VERSION")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(SERVICE_VERSION.to_string())),
environment: env::var("RUSTFS_OBSERVABILITY_ENVIRONMENT")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(ENVIRONMENT.to_string())),
logger_level: env::var("RUSTFS_OBSERVABILITY_LOGGER_LEVEL")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(DEFAULT_LOG_LEVEL.to_string())),
local_logging_enabled: env::var("RUSTFS_OBSERVABILITY_LOCAL_LOGGING_ENABLED")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(false)),
log_directory: env::var("RUSTFS_OBSERVABILITY_LOG_DIRECTORY")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(DEFAULT_LOG_DIR.to_string())),
log_filename: env::var("RUSTFS_OBSERVABILITY_LOG_FILENAME")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(DEFAULT_LOG_FILENAME.to_string())),
log_rotation_size_mb: env::var("RUSTFS_OBSERVABILITY_LOG_ROTATION_SIZE_MB")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(DEFAULT_LOG_ROTATION_SIZE_MB)), // Default to 100 MB
log_rotation_time: env::var("RUSTFS_OBSERVABILITY_LOG_ROTATION_TIME")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(DEFAULT_LOG_ROTATION_TIME.to_string())), // Default to "Day"
log_keep_files: env::var("RUSTFS_OBSERVABILITY_LOG_KEEP_FILES")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(DEFAULT_LOG_KEEP_FILES)), // Default to keeping 30 log files
}
}
/// Create a new instance of OtelConfig with default values
///
/// # Returns
/// A new instance of OtelConfig
pub fn new() -> Self {
extract_otel_config_from_env()
Self::extract_otel_config_from_env(None)
}
}
@@ -128,6 +164,12 @@ pub struct WebhookSinkConfig {
impl WebhookSinkConfig {
pub fn new() -> Self {
Self::default()
}
}
impl Default for WebhookSinkConfig {
fn default() -> Self {
Self {
endpoint: env::var("RUSTFS_SINKS_WEBHOOK_ENDPOINT")
.ok()
@@ -143,12 +185,6 @@ impl WebhookSinkConfig {
}
}
impl Default for WebhookSinkConfig {
fn default() -> Self {
Self::new()
}
}
/// File Sink Configuration - Add buffering parameters
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct FileSinkConfig {
@@ -166,7 +202,6 @@ impl FileSinkConfig {
eprintln!("Failed to create log directory: {}", e);
return "rustfs/rustfs.log".to_string();
}
println!("Using log directory: {:?}", temp_dir);
temp_dir
.join("rustfs.log")
.to_str()
@@ -179,9 +214,18 @@ impl FileSinkConfig {
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(Self::get_default_log_path),
buffer_size: Some(8192),
flush_interval_ms: Some(1000),
flush_threshold: Some(100),
buffer_size: env::var("RUSTFS_SINKS_FILE_BUFFER_SIZE")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(8192)),
flush_interval_ms: env::var("RUSTFS_SINKS_FILE_FLUSH_INTERVAL_MS")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(1000)),
flush_threshold: env::var("RUSTFS_SINKS_FILE_FLUSH_THRESHOLD")
.ok()
.and_then(|v| v.parse().ok())
.or(Some(100)),
}
}
}
@@ -266,6 +310,14 @@ impl AppConfig {
logger: Some(LoggerConfig::default()),
}
}
pub fn new_with_endpoint(endpoint: Option<String>) -> Self {
Self {
observability: OtelConfig::extract_otel_config_from_env(endpoint),
sinks: vec![SinkConfig::new()],
logger: Some(LoggerConfig::new()),
}
}
}
// implement default for AppConfig
@@ -275,9 +327,6 @@ impl Default for AppConfig {
}
}
/// Default configuration file name
const DEFAULT_CONFIG_FILE: &str = "obs";
/// Loading the configuration file
/// Supports TOML, YAML and .env formats, read in order by priority
///
@@ -294,54 +343,6 @@ const DEFAULT_CONFIG_FILE: &str = "obs";
///
/// let config = load_config(None);
/// ```
pub fn load_config(config_dir: Option<String>) -> AppConfig {
let config_dir = if let Some(path) = config_dir {
// If a path is provided, check if it's empty
if path.is_empty() {
// If empty, use the default config file name
DEFAULT_CONFIG_FILE.to_string()
} else {
// Use the provided path
let path = std::path::Path::new(&path);
if path.extension().is_some() {
// If path has extension, use it as is (extension will be added by Config::builder)
path.with_extension("").to_string_lossy().into_owned()
} else {
// If path is a directory, append the default config file name
path.to_string_lossy().into_owned()
}
}
} else {
// If no path provided, use current directory + default config file
match env::current_dir() {
Ok(dir) => {
println!("Current directory: {:?}", dir);
dir.join(DEFAULT_CONFIG_FILE).to_string_lossy().into_owned()
}
Err(_) => {
eprintln!("Warning: Failed to get current directory, using default config file");
DEFAULT_CONFIG_FILE.to_string()
}
}
};
// Log using proper logging instead of println when possible
println!("Using config file base: {}", config_dir);
let app_config = Config::builder()
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false))
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false))
.build()
.unwrap_or_default();
match app_config.try_deserialize::<AppConfig>() {
Ok(app_config) => {
println!("Parsed AppConfig: {:?}", app_config);
app_config
}
Err(e) => {
println!("Failed to deserialize config: {}", e);
AppConfig::default()
}
}
pub fn load_config(config: Option<String>) -> AppConfig {
AppConfig::new_with_endpoint(config)
}
crates/obs/src/telemetry.rs
+300 -136
View File
@@ -1,4 +1,7 @@
// Added flexi_logger related dependencies
use crate::OtelConfig;
use flexi_logger::{style, Age, Cleanup, Criterion, DeferredNow, FileSpec, LogSpecification, Naming, Record, WriteMode};
use nu_ansi_term::Color;
use opentelemetry::trace::TracerProvider;
use opentelemetry::{global, KeyValue};
use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
@@ -13,7 +16,9 @@ use opentelemetry_semantic_conventions::{
attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_VERSION as OTEL_SERVICE_VERSION},
SCHEMA_URL,
};
use rustfs_config::{APP_NAME, DEFAULT_LOG_LEVEL, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT};
use rustfs_config::{
APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_LEVEL, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT,
};
use rustfs_utils::get_local_ip_with_default;
use smallvec::SmallVec;
use std::borrow::Cow;
@@ -47,23 +52,44 @@ use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilte
/// // When it's dropped, all telemetry components are properly shut down
/// drop(otel_guard);
/// ```
#[derive(Debug)]
// Implement Debug trait correctly, rather than using derive, as some fields may not have implemented Debug
pub struct OtelGuard {
tracer_provider: SdkTracerProvider,
meter_provider: SdkMeterProvider,
logger_provider: SdkLoggerProvider,
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>,
}
// Implement debug manually and avoid relying on all fields to implement debug
impl std::fmt::Debug for OtelGuard {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OtelGuard")
.field("tracer_provider", &self.tracer_provider.is_some())
.field("meter_provider", &self.meter_provider.is_some())
.field("logger_provider", &self.logger_provider.is_some())
.field("_flexi_logger_handles", &self._flexi_logger_handles.is_some())
.finish()
}
}
impl Drop for OtelGuard {
fn drop(&mut self) {
if let Err(err) = self.tracer_provider.shutdown() {
eprintln!("Tracer shutdown error: {:?}", err);
if let Some(provider) = self.tracer_provider.take() {
if let Err(err) = provider.shutdown() {
eprintln!("Tracer shutdown error: {:?}", err);
}
}
if let Err(err) = self.meter_provider.shutdown() {
eprintln!("Meter shutdown error: {:?}", err);
if let Some(provider) = self.meter_provider.take() {
if let Err(err) = provider.shutdown() {
eprintln!("Meter shutdown error: {:?}", err);
}
}
if let Err(err) = self.logger_provider.shutdown() {
eprintln!("Logger shutdown error: {:?}", err);
if let Some(provider) = self.logger_provider.take() {
if let Err(err) = provider.shutdown() {
eprintln!("Logger shutdown error: {:?}", err);
}
}
}
}
@@ -106,156 +132,247 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
let service_name = config.service_name.as_deref().unwrap_or(APP_NAME);
let environment = config.environment.as_deref().unwrap_or(ENVIRONMENT);
// Pre-create resource objects to avoid repeated construction
let res = resource(config);
// Configure flexi_logger to cut by time and size
let mut flexi_logger_handle = None;
if !endpoint.is_empty() {
// Pre-create resource objects to avoid repeated construction
let res = resource(config);
// initialize tracer provider
let tracer_provider = {
let sample_ratio = config.sample_ratio.unwrap_or(SAMPLE_RATIO);
let sampler = if sample_ratio > 0.0 && sample_ratio < 1.0 {
Sampler::TraceIdRatioBased(sample_ratio)
} else {
Sampler::AlwaysOn
// initialize tracer provider
let tracer_provider = {
let sample_ratio = config.sample_ratio.unwrap_or(SAMPLE_RATIO);
let sampler = if sample_ratio > 0.0 && sample_ratio < 1.0 {
Sampler::TraceIdRatioBased(sample_ratio)
} else {
Sampler::AlwaysOn
};
let builder = SdkTracerProvider::builder()
.with_sampler(sampler)
.with_id_generator(RandomIdGenerator::default())
.with_resource(res.clone());
let tracer_provider = if endpoint.is_empty() {
builder
.with_batch_exporter(opentelemetry_stdout::SpanExporter::default())
.build()
} else {
let exporter = opentelemetry_otlp::SpanExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.build()
.unwrap();
let builder = if use_stdout {
builder
.with_batch_exporter(exporter)
.with_batch_exporter(opentelemetry_stdout::SpanExporter::default())
} else {
builder.with_batch_exporter(exporter)
};
builder.build()
};
global::set_tracer_provider(tracer_provider.clone());
tracer_provider
};
let builder = SdkTracerProvider::builder()
.with_sampler(sampler)
.with_id_generator(RandomIdGenerator::default())
.with_resource(res.clone());
// initialize meter provider
let meter_provider = {
let mut builder = MeterProviderBuilder::default().with_resource(res.clone());
let tracer_provider = if endpoint.is_empty() {
builder
.with_batch_exporter(opentelemetry_stdout::SpanExporter::default())
.build()
} else {
let exporter = opentelemetry_otlp::SpanExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.build()
.unwrap();
let builder = if use_stdout {
builder
.with_batch_exporter(exporter)
.with_batch_exporter(opentelemetry_stdout::SpanExporter::default())
if endpoint.is_empty() {
builder = builder.with_reader(create_periodic_reader(meter_interval));
} else {
builder.with_batch_exporter(exporter)
};
let exporter = opentelemetry_otlp::MetricExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.with_temporality(opentelemetry_sdk::metrics::Temporality::default())
.build()
.unwrap();
builder = builder.with_reader(
PeriodicReader::builder(exporter)
.with_interval(std::time::Duration::from_secs(meter_interval))
.build(),
);
if use_stdout {
builder = builder.with_reader(create_periodic_reader(meter_interval));
}
}
let meter_provider = builder.build();
global::set_meter_provider(meter_provider.clone());
meter_provider
};
// initialize logger provider
let logger_provider = {
let mut builder = SdkLoggerProvider::builder().with_resource(res);
if endpoint.is_empty() {
builder = builder.with_batch_exporter(opentelemetry_stdout::LogExporter::default());
} else {
let exporter = opentelemetry_otlp::LogExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.build()
.unwrap();
builder = builder.with_batch_exporter(exporter);
if use_stdout {
builder = builder.with_batch_exporter(opentelemetry_stdout::LogExporter::default());
}
}
builder.build()
};
global::set_tracer_provider(tracer_provider.clone());
tracer_provider
};
// configuring tracing
{
// configure the formatting layer
let fmt_layer = {
let enable_color = std::io::stdout().is_terminal();
let mut layer = tracing_subscriber::fmt::layer()
.with_target(true)
.with_ansi(enable_color)
.with_thread_names(true)
.with_thread_ids(true)
.with_file(true)
.with_line_number(true);
// initialize meter provider
let meter_provider = {
let mut builder = MeterProviderBuilder::default().with_resource(res.clone());
// Only add full span events tracking in the development environment
if environment != ENVIRONMENT {
layer = layer.with_span_events(FmtSpan::FULL);
}
if endpoint.is_empty() {
builder = builder.with_reader(create_periodic_reader(meter_interval));
} else {
let exporter = opentelemetry_otlp::MetricExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.with_temporality(opentelemetry_sdk::metrics::Temporality::default())
.build()
.unwrap();
layer.with_filter(build_env_filter(logger_level, None))
};
builder = builder.with_reader(
PeriodicReader::builder(exporter)
.with_interval(std::time::Duration::from_secs(meter_interval))
.build(),
);
let filter = build_env_filter(logger_level, None);
let otel_filter = build_env_filter(logger_level, None);
let otel_layer = OpenTelemetryTracingBridge::new(&logger_provider).with_filter(otel_filter);
let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string());
if use_stdout {
builder = builder.with_reader(create_periodic_reader(meter_interval));
// Configure registry to avoid repeated calls to filter methods
tracing_subscriber::registry()
.with(filter)
.with(ErrorLayer::default())
.with(if config.local_logging_enabled.unwrap_or(false) {
Some(fmt_layer)
} else {
None
})
.with(OpenTelemetryLayer::new(tracer))
.with(otel_layer)
.with(MetricsLayer::new(meter_provider.clone()))
.init();
if !endpoint.is_empty() {
info!(
"OpenTelemetry telemetry initialized with OTLP endpoint: {}, logger_level: {},RUST_LOG env: {}",
endpoint,
logger_level,
std::env::var("RUST_LOG").unwrap_or_else(|_| "Not set".to_string())
);
}
}
let meter_provider = builder.build();
global::set_meter_provider(meter_provider.clone());
meter_provider
};
// initialize logger provider
let logger_provider = {
let mut builder = SdkLoggerProvider::builder().with_resource(res);
if endpoint.is_empty() {
builder = builder.with_batch_exporter(opentelemetry_stdout::LogExporter::default());
} else {
let exporter = opentelemetry_otlp::LogExporter::builder()
.with_tonic()
.with_endpoint(endpoint)
.build()
.unwrap();
builder = builder.with_batch_exporter(exporter);
if use_stdout {
builder = builder.with_batch_exporter(opentelemetry_stdout::LogExporter::default());
}
OtelGuard {
tracer_provider: Some(tracer_provider),
meter_provider: Some(meter_provider),
logger_provider: Some(logger_provider),
_flexi_logger_handles: flexi_logger_handle,
}
} else {
// Obtain the log directory and file name configuration
let log_directory = config.log_directory.as_deref().unwrap_or("logs");
let log_filename = config.log_filename.as_deref().unwrap_or(service_name);
builder.build()
};
// configuring tracing
{
// configure the formatting layer
let fmt_layer = {
let enable_color = std::io::stdout().is_terminal();
let mut layer = tracing_subscriber::fmt::layer()
.with_target(true)
.with_ansi(enable_color)
.with_thread_names(true)
.with_thread_ids(true)
.with_file(true)
.with_line_number(true);
// Only add full span events tracking in the development environment
if environment != ENVIRONMENT {
layer = layer.with_span_events(FmtSpan::FULL);
// Build log cutting conditions
let rotation_criterion = match (config.log_rotation_time.as_deref(), config.log_rotation_size_mb) {
// Cut by time and size at the same time
(Some(time), Some(size)) => {
let age = match time.to_lowercase().as_str() {
"hour" => Age::Hour,
"day" => Age::Day,
"minute" => Age::Minute,
"second" => Age::Second,
_ => Age::Day, // The default is by day
};
Criterion::AgeOrSize(age, size * 1024 * 1024) // Convert to bytes
}
layer.with_filter(build_env_filter(logger_level, None))
// Cut by time only
(Some(time), None) => {
let age = match time.to_lowercase().as_str() {
"hour" => Age::Hour,
"day" => Age::Day,
"minute" => Age::Minute,
"second" => Age::Second,
_ => Age::Day, // The default is by day
};
Criterion::Age(age)
}
// Cut by size only
(None, Some(size)) => {
Criterion::Size(size * 1024 * 1024) // Convert to bytes
}
// By default, it is cut by the day
_ => Criterion::Age(Age::Day),
};
let filter = build_env_filter(logger_level, None);
let otel_filter = build_env_filter(logger_level, None);
let otel_layer = OpenTelemetryTracingBridge::new(&logger_provider).with_filter(otel_filter);
let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string());
// The number of log files retained
let keep_files = config.log_keep_files.unwrap_or(DEFAULT_LOG_KEEP_FILES);
// Configure registry to avoid repeated calls to filter methods
tracing_subscriber::registry()
.with(filter)
.with(ErrorLayer::default())
.with(if config.local_logging_enabled.unwrap_or(false) {
Some(fmt_layer)
} else {
None
})
.with(OpenTelemetryLayer::new(tracer))
.with(otel_layer)
.with(MetricsLayer::new(meter_provider.clone()))
.init();
// Parsing the log level
let log_spec = LogSpecification::parse(logger_level).unwrap_or(LogSpecification::info());
if !endpoint.is_empty() {
info!(
"OpenTelemetry telemetry initialized with OTLP endpoint: {}, logger_level: {},RUST_LOG env: {}",
endpoint,
logger_level,
std::env::var("RUST_LOG").unwrap_or_else(|_| "Not set".to_string())
);
// Configure the flexi_logger
let flexi_logger_result = flexi_logger::Logger::with(log_spec)
.log_to_file(
FileSpec::default()
.directory(log_directory)
.basename(log_filename)
.suffix("log"),
)
.rotate(rotation_criterion, Naming::Timestamps, Cleanup::KeepLogFiles(keep_files.into()))
.format_for_files(format_for_file) // Add a custom formatting function for file output
.duplicate_to_stdout(flexi_logger::Duplicate::Info)
.format_for_stdout(format_with_color) // Add a custom formatting function for terminal output
.write_mode(WriteMode::Async)
.start();
if let Ok(logger) = flexi_logger_result {
// Save the logger handle to keep the logging
flexi_logger_handle = Some(logger);
info!("Flexi logger initialized with file logging to {}/{}.log", log_directory, log_filename);
// Log logging of log cutting conditions
match (config.log_rotation_time.as_deref(), config.log_rotation_size_mb) {
(Some(time), Some(size)) => info!(
"Log rotation configured for: every {} or when size exceeds {}MB, keeping {} files",
time, size, keep_files
),
(Some(time), None) => info!("Log rotation configured for: every {}, keeping {} files", time, keep_files),
(None, Some(size)) => {
info!("Log rotation configured for: when size exceeds {}MB, keeping {} files", size, keep_files)
}
_ => info!("Log rotation configured for: daily, keeping {} files", keep_files),
}
} else {
eprintln!("Failed to initialize flexi_logger: {:?}", flexi_logger_result.err());
}
}
OtelGuard {
tracer_provider,
meter_provider,
logger_provider,
OtelGuard {
tracer_provider: None,
meter_provider: None,
logger_provider: None,
_flexi_logger_handles: flexi_logger_handle,
}
}
}
@@ -271,3 +388,50 @@ fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilte
filter
}
/// Custom Log Formatter Function - Terminal Output (with Color)
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());
write!(
w,
"{} {} [{}] [{}:{}] [{} {}] {}\n",
now.now().format("%Y-%m-%d %H:%M:%S%.6f"),
level_style.paint(level.to_string()),
Color::Magenta.paint(record.target()),
Color::Blue.paint(record.file().unwrap_or("unknown")),
Color::Blue.paint(record.line().unwrap_or(0).to_string()),
Color::Green.paint(thread_name),
Color::Green.paint(thread_id),
record.args()
)
}
/// Custom Log Formatter - File Output (No Color)
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());
write!(
w,
"{} {} [{}] [{}:{}] [{} {}] {}\n",
now.now().format("%Y-%m-%d %H:%M:%S%.6f"),
level.to_string(),
record.target(),
record.file().unwrap_or("unknown"),
record.line().unwrap_or(0),
thread_name,
thread_id,
record.args()
)
}