mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-31 09:18:28 +00:00
improve code for request and telemetry
This commit is contained in:
@@ -210,7 +210,8 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
|
|||||||
.with_thread_names(true)
|
.with_thread_names(true)
|
||||||
.with_thread_ids(true)
|
.with_thread_ids(true)
|
||||||
.with_file(true)
|
.with_file(true)
|
||||||
.with_line_number(true);
|
.with_line_number(true)
|
||||||
|
.with_filter(build_env_filter(logger_level, None));
|
||||||
|
|
||||||
let filter = build_env_filter(logger_level, None);
|
let filter = build_env_filter(logger_level, None);
|
||||||
let otel_filter = build_env_filter(logger_level, None);
|
let otel_filter = build_env_filter(logger_level, None);
|
||||||
@@ -231,16 +232,13 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
|
|||||||
.with(MetricsLayer::new(meter_provider.clone()))
|
.with(MetricsLayer::new(meter_provider.clone()))
|
||||||
.init();
|
.init();
|
||||||
info!("Telemetry logging enabled: {:?}", config.local_logging_enabled);
|
info!("Telemetry logging enabled: {:?}", config.local_logging_enabled);
|
||||||
// if config.local_logging_enabled.unwrap_or(false) {
|
|
||||||
// registry.with(fmt_layer).init();
|
|
||||||
// } else {
|
|
||||||
// registry.init();
|
|
||||||
// }
|
|
||||||
|
|
||||||
if !endpoint.is_empty() {
|
if !endpoint.is_empty() {
|
||||||
info!(
|
info!(
|
||||||
"OpenTelemetry telemetry initialized with OTLP endpoint: {}, logger_level: {}",
|
"OpenTelemetry telemetry initialized with OTLP endpoint: {}, logger_level: {},RUST_LOG env: {}",
|
||||||
endpoint, logger_level
|
endpoint,
|
||||||
|
logger_level,
|
||||||
|
std::env::var("RUST_LOG").unwrap_or_else(|_| "未设置".to_string())
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -255,7 +253,6 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
|
|||||||
fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter {
|
fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter {
|
||||||
let level = default_level.unwrap_or(logger_level);
|
let level = default_level.unwrap_or(logger_level);
|
||||||
let mut filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level));
|
let mut filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level));
|
||||||
|
|
||||||
if !matches!(logger_level, "trace" | "debug") {
|
if !matches!(logger_level, "trace" | "debug") {
|
||||||
let directives: SmallVec<[&str; 5]> = smallvec::smallvec!["hyper", "tonic", "h2", "reqwest", "tower"];
|
let directives: SmallVec<[&str; 5]> = smallvec::smallvec!["hyper", "tonic", "h2", "reqwest", "tower"];
|
||||||
for directive in directives {
|
for directive in directives {
|
||||||
|
|||||||
+19
-16
@@ -62,8 +62,8 @@ use tokio_rustls::TlsAcceptor;
|
|||||||
use tonic::{metadata::MetadataValue, Request, Status};
|
use tonic::{metadata::MetadataValue, Request, Status};
|
||||||
use tower_http::cors::CorsLayer;
|
use tower_http::cors::CorsLayer;
|
||||||
use tower_http::trace::TraceLayer;
|
use tower_http::trace::TraceLayer;
|
||||||
use tracing::Span;
|
use tracing::{debug, error, info, warn};
|
||||||
use tracing::{debug, error, info, info_span, warn};
|
use tracing::{instrument, Span};
|
||||||
|
|
||||||
#[cfg(all(target_os = "linux", target_env = "gnu"))]
|
#[cfg(all(target_os = "linux", target_env = "gnu"))]
|
||||||
#[global_allocator]
|
#[global_allocator]
|
||||||
@@ -77,7 +77,7 @@ fn check_auth(req: Request<()>) -> Result<Request<()>, Status> {
|
|||||||
_ => Err(Status::unauthenticated("No valid auth token")),
|
_ => Err(Status::unauthenticated("No valid auth token")),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
#[instrument]
|
||||||
fn print_server_info() {
|
fn print_server_info() {
|
||||||
let cfg = CONSOLE_CONFIG.get().unwrap();
|
let cfg = CONSOLE_CONFIG.get().unwrap();
|
||||||
let current_year = chrono::Utc::now().year();
|
let current_year = chrono::Utc::now().year();
|
||||||
@@ -95,8 +95,7 @@ async fn main() -> Result<()> {
|
|||||||
// Parse the obtained parameters
|
// Parse the obtained parameters
|
||||||
let opt = config::Opt::parse();
|
let opt = config::Opt::parse();
|
||||||
|
|
||||||
// config::init_config(opt.clone());
|
// Initialize the configuration
|
||||||
|
|
||||||
init_license(opt.license.clone());
|
init_license(opt.license.clone());
|
||||||
|
|
||||||
// Load the configuration file
|
// Load the configuration file
|
||||||
@@ -108,8 +107,13 @@ async fn main() -> Result<()> {
|
|||||||
// Store in global storage
|
// Store in global storage
|
||||||
set_global_guard(guard)?;
|
set_global_guard(guard)?;
|
||||||
|
|
||||||
|
// Run parameters
|
||||||
|
run(opt).await
|
||||||
|
}
|
||||||
|
|
||||||
|
#[instrument]
|
||||||
|
async fn init_event_notifier(notifier_config: Option<String>) {
|
||||||
// Initialize event notifier
|
// Initialize event notifier
|
||||||
let notifier_config = opt.clone().event_config;
|
|
||||||
if notifier_config.is_some() {
|
if notifier_config.is_some() {
|
||||||
info!("event_config is not empty");
|
info!("event_config is not empty");
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
@@ -124,18 +128,16 @@ async fn main() -> Result<()> {
|
|||||||
} else {
|
} else {
|
||||||
info!("event_config is empty");
|
info!("event_config is empty");
|
||||||
}
|
}
|
||||||
|
|
||||||
// Run parameters
|
|
||||||
run(opt).await
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// #[tokio::main]
|
#[instrument(skip(opt))]
|
||||||
async fn run(opt: config::Opt) -> Result<()> {
|
async fn run(opt: config::Opt) -> Result<()> {
|
||||||
let span = info_span!("trace-main-run");
|
|
||||||
let _enter = span.enter();
|
|
||||||
|
|
||||||
debug!("opt: {:?}", &opt);
|
debug!("opt: {:?}", &opt);
|
||||||
|
|
||||||
|
// Initialize event notifier
|
||||||
|
let notifier_config = opt.event_config;
|
||||||
|
init_event_notifier(notifier_config).await;
|
||||||
|
|
||||||
let server_addr = net::parse_and_resolve_address(opt.address.as_str())?;
|
let server_addr = net::parse_and_resolve_address(opt.address.as_str())?;
|
||||||
let server_port = server_addr.port();
|
let server_port = server_addr.port();
|
||||||
let server_address = server_addr.to_string();
|
let server_address = server_addr.to_string();
|
||||||
@@ -175,7 +177,7 @@ async fn run(opt: config::Opt) -> Result<()> {
|
|||||||
// Detailed endpoint information (showing all API endpoints)
|
// Detailed endpoint information (showing all API endpoints)
|
||||||
let api_endpoints = format!("http://{}:{}", local_ip, server_port);
|
let api_endpoints = format!("http://{}:{}", local_ip, server_port);
|
||||||
let localhost_endpoint = format!("http://127.0.0.1:{}", server_port);
|
let localhost_endpoint = format!("http://127.0.0.1:{}", server_port);
|
||||||
info!("API: {} {}", api_endpoints, localhost_endpoint);
|
info!(" API: {} {}", api_endpoints, localhost_endpoint);
|
||||||
info!(" RootUser: {}", opt.access_key.clone());
|
info!(" RootUser: {}", opt.access_key.clone());
|
||||||
info!(" RootPass: {}", opt.secret_key.clone());
|
info!(" RootPass: {}", opt.secret_key.clone());
|
||||||
if DEFAULT_ACCESS_KEY.eq(&opt.access_key) && DEFAULT_SECRET_KEY.eq(&opt.secret_key) {
|
if DEFAULT_ACCESS_KEY.eq(&opt.access_key) && DEFAULT_SECRET_KEY.eq(&opt.secret_key) {
|
||||||
@@ -339,7 +341,7 @@ async fn run(opt: config::Opt) -> Result<()> {
|
|||||||
.layer(
|
.layer(
|
||||||
TraceLayer::new_for_http()
|
TraceLayer::new_for_http()
|
||||||
.make_span_with(|request: &HttpRequest<_>| {
|
.make_span_with(|request: &HttpRequest<_>| {
|
||||||
let span = tracing::debug_span!("http-request",
|
let span = tracing::info_span!("http-request",
|
||||||
status_code = tracing::field::Empty,
|
status_code = tracing::field::Empty,
|
||||||
method = %request.method(),
|
method = %request.method(),
|
||||||
uri = %request.uri(),
|
uri = %request.uri(),
|
||||||
@@ -368,13 +370,14 @@ async fn run(opt: config::Opt) -> Result<()> {
|
|||||||
debug!("http response generated in {:?}", latency)
|
debug!("http response generated in {:?}", latency)
|
||||||
})
|
})
|
||||||
.on_body_chunk(|chunk: &Bytes, latency: Duration, _span: &Span| {
|
.on_body_chunk(|chunk: &Bytes, latency: Duration, _span: &Span| {
|
||||||
info!(histogram.request.body.len = chunk.len(), "histogram request body lenght",);
|
info!(histogram.request.body.len = chunk.len(), "histogram request body length",);
|
||||||
debug!("http body sending {} bytes in {:?}", chunk.len(), latency)
|
debug!("http body sending {} bytes in {:?}", chunk.len(), latency)
|
||||||
})
|
})
|
||||||
.on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, _span: &Span| {
|
.on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, _span: &Span| {
|
||||||
debug!("http stream closed after {:?}", stream_duration)
|
debug!("http stream closed after {:?}", stream_duration)
|
||||||
})
|
})
|
||||||
.on_failure(|_error, latency: Duration, _span: &Span| {
|
.on_failure(|_error, latency: Duration, _span: &Span| {
|
||||||
|
info!(counter.rustfs_api_requests_failure_total = 1_u64, "handle request api failure total");
|
||||||
debug!("http request failure error: {:?} in {:?}", _error, latency)
|
debug!("http request failure error: {:?} in {:?}", _error, latency)
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user