mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-18 18:46:17 +00:00
improve code for observability
This commit is contained in:
houseme
@@ -1,319 +0,0 @@
|
||||
#[cfg(feature = "audit-kafka")]
|
||||
use rdkafka::{
|
||||
producer::{FutureProducer, FutureRecord},
|
||||
ClientConfig,
|
||||
};
|
||||
|
||||
#[cfg(feature = "audit-webhook")]
|
||||
use reqwest::Client;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
/// AuditEntry is a struct that represents an audit entry
|
||||
/// that can be logged
|
||||
/// # Fields
|
||||
/// * `version` - The version of the audit entry
|
||||
/// * `event_type` - The type of event that occurred
|
||||
/// * `bucket` - The bucket that was accessed
|
||||
/// * `object` - The object that was accessed
|
||||
/// * `user` - The user that accessed the object
|
||||
/// * `time` - The time the event occurred
|
||||
/// * `user_agent` - The user agent that accessed the object
|
||||
/// * `span_id` - The span ID of the event
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::AuditEntry;
|
||||
/// let entry = AuditEntry {
|
||||
/// version: "1.0".to_string(),
|
||||
/// event_type: "read".to_string(),
|
||||
/// bucket: "bucket".to_string(),
|
||||
/// object: "object".to_string(),
|
||||
/// user: "user".to_string(),
|
||||
/// time: "time".to_string(),
|
||||
/// user_agent: "user_agent".to_string(),
|
||||
/// span_id: "span_id".to_string(),
|
||||
/// };
|
||||
/// ```
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
pub struct AuditEntry {
|
||||
pub version: String,
|
||||
pub event_type: String,
|
||||
pub bucket: String,
|
||||
pub object: String,
|
||||
pub user: String,
|
||||
pub time: String,
|
||||
pub user_agent: String,
|
||||
pub span_id: String,
|
||||
}
|
||||
|
||||
/// AuditTarget is a trait that defines the interface for audit targets
|
||||
/// that can receive audit entries
|
||||
pub trait AuditTarget: Send + Sync {
|
||||
fn send(&self, entry: AuditEntry);
|
||||
}
|
||||
|
||||
/// FileAuditTarget is an audit target that logs audit entries to a file
|
||||
pub struct FileAuditTarget;
|
||||
|
||||
impl AuditTarget for FileAuditTarget {
|
||||
/// Send an audit entry to a file
|
||||
/// # Arguments
|
||||
/// * `entry` - The audit entry to send
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::{AuditEntry, AuditTarget, FileAuditTarget};
|
||||
/// let entry = AuditEntry {
|
||||
/// version: "1.0".to_string(),
|
||||
/// event_type: "read".to_string(),
|
||||
/// bucket: "bucket".to_string(),
|
||||
/// object: "object".to_string(),
|
||||
/// user: "user".to_string(),
|
||||
/// time: "time".to_string(),
|
||||
/// user_agent: "user_agent".to_string(),
|
||||
/// span_id: "span_id".to_string(),
|
||||
/// };
|
||||
/// FileAuditTarget.send(entry);
|
||||
/// ```
|
||||
fn send(&self, entry: AuditEntry) {
|
||||
println!("File audit: {:?}", entry);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "audit-webhook")]
|
||||
/// Webhook audit objectives
|
||||
/// #Arguments
|
||||
/// * `client` - The reqwest client
|
||||
/// * `url` - The URL of the webhook
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::WebhookAuditTarget;
|
||||
/// let target = WebhookAuditTarget::new("http://localhost:8080");
|
||||
/// ```
|
||||
pub struct WebhookAuditTarget {
|
||||
client: Client,
|
||||
url: String,
|
||||
}
|
||||
|
||||
#[cfg(feature = "audit-webhook")]
|
||||
impl WebhookAuditTarget {
|
||||
pub fn new(url: &str) -> Self {
|
||||
Self {
|
||||
client: Client::new(),
|
||||
url: url.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "audit-webhook")]
|
||||
impl AuditTarget for WebhookAuditTarget {
|
||||
fn send(&self, entry: AuditEntry) {
|
||||
let client = self.client.clone();
|
||||
let url = self.url.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = client.post(&url).json(&entry).send().await {
|
||||
eprintln!("Failed to send to Webhook: {:?}", e);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "audit-kafka")]
|
||||
/// Kafka audit objectives
|
||||
/// # Arguments
|
||||
/// * `producer` - The Kafka producer
|
||||
/// * `topic` - The Kafka topic
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::KafkaAuditTarget;
|
||||
/// let target = KafkaAuditTarget::new("localhost:9092", "rustfs-audit");
|
||||
/// ```
|
||||
/// # Note
|
||||
/// This feature requires the `rdkafka` crate
|
||||
/// # Example
|
||||
/// ```toml
|
||||
/// [dependencies]
|
||||
/// rdkafka = "0.26.0"
|
||||
/// rustfs_obs = { version = "0.1.0", features = ["audit-kafka"] }
|
||||
/// ```
|
||||
/// # Note
|
||||
/// The `rdkafka` crate requires the `librdkafka` library to be installed
|
||||
/// # Example
|
||||
/// ```sh
|
||||
/// sudo apt-get install librdkafka-dev
|
||||
/// ```
|
||||
/// # Note
|
||||
/// The `rdkafka` crate requires the `libssl-dev` and `pkg-config` packages to be installed
|
||||
/// # Example
|
||||
/// ```sh
|
||||
/// sudo apt-get install libssl-dev pkg-config
|
||||
/// ```
|
||||
/// # Note
|
||||
/// The `rdkafka` crate requires the `zlib1g-dev` package to be installed
|
||||
/// # Example
|
||||
/// ```sh
|
||||
/// sudo apt-get install zlib1g-dev
|
||||
/// ```
|
||||
/// # Note
|
||||
/// The `rdkafka` crate requires the `zstd` package to be installed
|
||||
/// # Example
|
||||
/// ```sh
|
||||
/// sudo apt-get install zstd
|
||||
/// ```
|
||||
/// # Note
|
||||
/// The `rdkafka` crate requires the `lz4` package to be installed
|
||||
/// # Example
|
||||
/// ```sh
|
||||
/// sudo apt-get install lz4
|
||||
/// ```
|
||||
pub struct KafkaAuditTarget {
|
||||
producer: FutureProducer,
|
||||
topic: String,
|
||||
}
|
||||
|
||||
#[cfg(feature = "audit-kafka")]
|
||||
impl KafkaAuditTarget {
|
||||
pub fn new(brokers: &str, topic: &str) -> Self {
|
||||
let producer: FutureProducer = ClientConfig::new()
|
||||
.set("bootstrap.servers", brokers)
|
||||
.set("message.timeout.ms", "5000")
|
||||
.create()
|
||||
.expect("Kafka producer creation failed");
|
||||
Self {
|
||||
producer,
|
||||
topic: topic.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "audit-kafka")]
|
||||
impl AuditTarget for KafkaAuditTarget {
|
||||
fn send(&self, entry: AuditEntry) {
|
||||
let topic = self.topic.clone();
|
||||
let span_id = entry.span_id.clone();
|
||||
let payload = serde_json::to_string(&entry).unwrap();
|
||||
// let record = FutureRecord::to(&topic).payload(&payload).key(&span_id);
|
||||
tokio::spawn({
|
||||
// 在异步闭包内部创建 record
|
||||
let topic = topic;
|
||||
let payload = payload;
|
||||
let span_id = span_id;
|
||||
let producer = self.producer.clone();
|
||||
async move {
|
||||
let record = FutureRecord::to(&topic).payload(&payload).key(&span_id);
|
||||
if let Err(e) = producer.send(record, std::time::Duration::from_secs(0)).await {
|
||||
eprintln!("Failed to send to Kafka: {:?}", e);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
/// AuditLogger is a logger that logs audit entries
|
||||
/// to multiple targets
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::{AuditEntry, AuditLogger, FileAuditTarget};
|
||||
///
|
||||
/// #[tokio::main]
|
||||
/// async fn main() {
|
||||
/// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]);
|
||||
/// let entry = AuditEntry {
|
||||
/// version: "1.0".to_string(),
|
||||
/// event_type: "read".to_string(),
|
||||
/// bucket: "bucket".to_string(),
|
||||
/// object: "object".to_string(),
|
||||
/// user: "user".to_string(),
|
||||
/// time: "time".to_string(),
|
||||
/// user_agent: "user_agent".to_string(),
|
||||
/// span_id: "span_id".to_string(),
|
||||
/// };
|
||||
/// logger.log(entry).await;
|
||||
/// }
|
||||
/// ```
|
||||
|
||||
#[derive(Debug)]
|
||||
/// AuditLogger is a logger that logs audit entries
|
||||
/// to multiple targets
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::{AuditEntry, AuditLogger, FileAuditTarget};
|
||||
/// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]);
|
||||
/// ```
|
||||
/// # Note
|
||||
/// This feature requires the `tokio` crate
|
||||
/// # Example
|
||||
/// ```toml
|
||||
/// [dependencies]
|
||||
/// tokio = { version = "1", features = ["full"] }
|
||||
/// rustfs_obs = { version = "0.1.0"}
|
||||
/// ```
|
||||
/// # Note
|
||||
/// This feature requires the `serde` crate
|
||||
/// # Example
|
||||
/// ```toml
|
||||
/// [dependencies]
|
||||
/// serde = { version = "1", features = ["derive"] }
|
||||
/// rustfs_obs = { version = "0.1.0"}
|
||||
/// ```
|
||||
pub struct AuditLogger {
|
||||
tx: mpsc::Sender<AuditEntry>,
|
||||
}
|
||||
|
||||
impl AuditLogger {
|
||||
/// Create a new AuditLogger with the given targets
|
||||
/// that will receive audit entries
|
||||
/// # Arguments
|
||||
/// * `targets` - A vector of audit targets
|
||||
/// # Returns
|
||||
/// * An AuditLogger
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::{AuditLogger, AuditEntry, FileAuditTarget};
|
||||
///
|
||||
/// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]);
|
||||
/// ```
|
||||
pub fn new(targets: Vec<Box<dyn AuditTarget>>) -> Self {
|
||||
let (tx, mut rx) = mpsc::channel::<AuditEntry>(1000);
|
||||
tokio::spawn(async move {
|
||||
while let Some(entry) = rx.recv().await {
|
||||
for target in &targets {
|
||||
target.send(entry.clone());
|
||||
}
|
||||
}
|
||||
});
|
||||
Self { tx }
|
||||
}
|
||||
|
||||
/// Log an audit entry
|
||||
/// # Arguments
|
||||
/// * `entry` - The audit entry to log
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::{AuditEntry, AuditLogger, FileAuditTarget};
|
||||
///
|
||||
/// #[tokio::main]
|
||||
/// async fn main() {
|
||||
/// let logger = AuditLogger::new(vec![Box::new(FileAuditTarget)]);
|
||||
/// let entry = AuditEntry {
|
||||
/// version: "1.0".to_string(),
|
||||
/// event_type: "read".to_string(),
|
||||
/// bucket: "bucket".to_string(),
|
||||
/// object: "object".to_string(),
|
||||
/// user: "user".to_string(),
|
||||
/// time: "time".to_string(),
|
||||
/// user_agent: "user_agent".to_string(),
|
||||
/// span_id: "span_id".to_string(),
|
||||
/// };
|
||||
/// logger.log(entry).await;
|
||||
/// }
|
||||
/// ```
|
||||
pub async fn log(&self, entry: AuditEntry) {
|
||||
// 将日志消息记录到当前 Span
|
||||
tracing::Span::current()
|
||||
.record("log_message", &entry.bucket)
|
||||
.record("source", &entry.event_type);
|
||||
let _ = self.tx.send(entry).await;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
use config::{Config, File, FileFormat};
|
||||
use serde::Deserialize;
|
||||
use std::env;
|
||||
|
||||
/// OpenTelemetry Configuration
|
||||
#[derive(Debug, Deserialize, Clone, Default)]
|
||||
pub struct OtelConfig {
|
||||
pub endpoint: String,
|
||||
pub use_stdout: bool,
|
||||
pub sample_ratio: f64,
|
||||
pub meter_interval: u64,
|
||||
pub service_name: String,
|
||||
pub service_version: String,
|
||||
pub deployment_environment: String,
|
||||
}
|
||||
|
||||
/// Kafka Sink Configuration - Add batch parameters
|
||||
#[derive(Debug, Deserialize, Clone)]
|
||||
pub struct KafkaSinkConfig {
|
||||
pub enabled: bool,
|
||||
pub bootstrap_servers: String,
|
||||
pub topic: String,
|
||||
pub batch_size: Option<usize>, // Batch size, default 100
|
||||
pub batch_timeout_ms: Option<u64>, // Batch timeout time, default 1000ms
|
||||
}
|
||||
|
||||
/// Webhook Sink Configuration - Add Retry Parameters
|
||||
#[derive(Debug, Deserialize, Clone)]
|
||||
pub struct WebhookSinkConfig {
|
||||
pub enabled: bool,
|
||||
pub url: String,
|
||||
pub max_retries: Option<usize>, // Maximum number of retry times, default 3
|
||||
pub retry_delay_ms: Option<u64>, // Retry the delay cardinality, default 100ms
|
||||
}
|
||||
|
||||
/// File Sink Configuration - Add buffering parameters
|
||||
#[derive(Debug, Deserialize, Clone)]
|
||||
pub struct FileSinkConfig {
|
||||
pub enabled: bool,
|
||||
pub path: String,
|
||||
pub buffer_size: Option<usize>, // Write buffer size, default 8192
|
||||
pub flush_interval_ms: Option<u64>, // Refresh interval time, default 1000ms
|
||||
pub flush_threshold: Option<usize>, // Refresh threshold, default 100 logs
|
||||
}
|
||||
|
||||
/// Sink configuration collection
|
||||
#[derive(Debug, Deserialize, Clone)]
|
||||
pub struct SinkConfig {
|
||||
pub kafka: KafkaSinkConfig,
|
||||
pub webhook: WebhookSinkConfig,
|
||||
pub file: FileSinkConfig,
|
||||
}
|
||||
|
||||
///Logger Configuration
|
||||
#[derive(Debug, Deserialize, Clone)]
|
||||
pub struct LoggerConfig {
|
||||
pub queue_capacity: Option<usize>,
|
||||
}
|
||||
|
||||
/// Overall application configuration
|
||||
#[derive(Debug, Deserialize, Clone)]
|
||||
pub struct AppConfig {
|
||||
pub observability: OtelConfig,
|
||||
pub sinks: SinkConfig,
|
||||
pub logger: LoggerConfig,
|
||||
}
|
||||
|
||||
/// Loading the configuration file
|
||||
/// Supports TOML, YAML and .env formats, read in order by priority
|
||||
pub fn load_config(config_dir: Option<String>) -> AppConfig {
|
||||
let config_dir = config_dir.unwrap_or_else(|| {
|
||||
env::current_dir()
|
||||
.map(|path| path.to_string_lossy().to_string())
|
||||
.unwrap_or_else(|_| {
|
||||
eprintln!("Warning: Failed to get current directory, using empty path");
|
||||
String::new()
|
||||
})
|
||||
});
|
||||
|
||||
println!("config_dir: {}", config_dir);
|
||||
|
||||
let config = Config::builder()
|
||||
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml))
|
||||
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false))
|
||||
.add_source(config::Environment::with_prefix(""))
|
||||
.build()
|
||||
.unwrap();
|
||||
|
||||
config.try_deserialize().unwrap()
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde::de::Error;
|
||||
use serde::{Deserialize, Deserializer, Serialize, Serializer};
|
||||
use tracing_core::Level;
|
||||
|
||||
/// Wrapper for `tracing_core::Level` to implement `Serialize` and `Deserialize`
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SerializableLevel(pub Level);
|
||||
|
||||
impl From<Level> for SerializableLevel {
|
||||
fn from(level: Level) -> Self {
|
||||
SerializableLevel(level)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<SerializableLevel> for Level {
|
||||
fn from(serializable_level: SerializableLevel) -> Self {
|
||||
serializable_level.0
|
||||
}
|
||||
}
|
||||
|
||||
impl Serialize for SerializableLevel {
|
||||
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
serializer.serialize_str(self.0.as_str())
|
||||
}
|
||||
}
|
||||
|
||||
impl<'de> Deserialize<'de> for SerializableLevel {
|
||||
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let s = String::deserialize(deserializer)?;
|
||||
match s.as_str() {
|
||||
"TRACE" => Ok(SerializableLevel(Level::TRACE)),
|
||||
"DEBUG" => Ok(SerializableLevel(Level::DEBUG)),
|
||||
"INFO" => Ok(SerializableLevel(Level::INFO)),
|
||||
"WARN" => Ok(SerializableLevel(Level::WARN)),
|
||||
"ERROR" => Ok(SerializableLevel(Level::ERROR)),
|
||||
_ => Err(D::Error::custom("unknown log level")),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Server log entry structure
|
||||
#[derive(Debug, Serialize, Deserialize, Clone)]
|
||||
pub struct LogEntry {
|
||||
pub timestamp: DateTime<Utc>, // Log timestamp
|
||||
pub level: SerializableLevel, // Log Level
|
||||
pub message: String, // Log messages
|
||||
pub source: String, // Log source (such as module name)
|
||||
pub request_id: Option<String>, // Request ID (Common Server Fields)
|
||||
pub user_id: Option<String>, // User ID (Common Server Fields)
|
||||
pub fields: Vec<(String, String)>, // Attached fields (key value pairs)
|
||||
}
|
||||
|
||||
impl LogEntry {
|
||||
/// Create a new LogEntry
|
||||
pub fn new(
|
||||
level: Level,
|
||||
message: String,
|
||||
source: String,
|
||||
request_id: Option<String>,
|
||||
user_id: Option<String>,
|
||||
fields: Vec<(String, String)>,
|
||||
) -> Self {
|
||||
LogEntry {
|
||||
timestamp: Utc::now(),
|
||||
level: SerializableLevel::from(level),
|
||||
message,
|
||||
source,
|
||||
request_id,
|
||||
user_id,
|
||||
fields,
|
||||
}
|
||||
}
|
||||
}
|
||||
+26
-184
@@ -1,190 +1,32 @@
|
||||
//! Logging utilities
|
||||
|
||||
/// # obs
|
||||
///
|
||||
/// This crate provides utilities for logging.
|
||||
///
|
||||
/// # Examples
|
||||
/// ```
|
||||
/// use rustfs_obs::{log_info, log_error};
|
||||
///
|
||||
/// log_info("This is an informational message");
|
||||
/// log_error("This is an error message");
|
||||
/// ```
|
||||
#[cfg(feature = "audit-kafka")]
|
||||
pub use audit::KafkaAuditTarget;
|
||||
#[cfg(feature = "audit-webhook")]
|
||||
pub use audit::WebhookAuditTarget;
|
||||
pub use audit::{AuditEntry, AuditLogger, AuditTarget, FileAuditTarget};
|
||||
pub use logger::{log_debug, log_error, log_info};
|
||||
pub use telemetry::Telemetry;
|
||||
|
||||
mod audit;
|
||||
/// `obs` is a logging and observability library for Rust.
|
||||
/// It provides a simple and easy-to-use interface for logging and observability.
|
||||
/// It is built on top of the `log` crate and `opentelemetry` crate.
|
||||
mod config;
|
||||
mod entry;
|
||||
mod logger;
|
||||
mod sink;
|
||||
mod telemetry;
|
||||
mod utils;
|
||||
mod worker;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::{log_info, AuditEntry, AuditLogger, AuditTarget, FileAuditTarget, Telemetry};
|
||||
use chrono::Utc;
|
||||
use opentelemetry::global;
|
||||
use opentelemetry::trace::{TraceContextExt, Tracer};
|
||||
use std::time::{Duration, SystemTime};
|
||||
use tracing::{instrument, Span};
|
||||
use tracing_opentelemetry::OpenTelemetrySpanExt;
|
||||
pub use config::load_config;
|
||||
pub use config::{AppConfig, OtelConfig};
|
||||
pub use entry::{LogEntry, SerializableLevel};
|
||||
pub use logger::start_logger;
|
||||
pub use logger::{LogError, Logger};
|
||||
pub use sink::Sink;
|
||||
pub use telemetry::init_telemetry;
|
||||
pub use utils::{get_local_ip, get_local_ip_with_default};
|
||||
pub use worker::start_worker;
|
||||
|
||||
#[instrument(fields(bucket, object, user))]
|
||||
async fn put_object(audit_logger: &AuditLogger, bucket: String, object: String, user: String) {
|
||||
let start_time = SystemTime::now();
|
||||
log_info("Starting PUT operation");
|
||||
|
||||
// Simulate the operation
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
// Record Metrics
|
||||
let meter = global::meter("rustfs.rs");
|
||||
let request_duration = meter.f64_histogram("s3_request_duration_seconds").build();
|
||||
request_duration.record(
|
||||
start_time.elapsed().unwrap().as_secs_f64(),
|
||||
&[opentelemetry::KeyValue::new("operation", "put_object")],
|
||||
);
|
||||
|
||||
// Gets the current span
|
||||
let span = Span::current();
|
||||
|
||||
// Use 'OpenTelemetrySpanExt' to get 'SpanContext'
|
||||
let span_context = span.context(); // Get context via OpenTelemetrySpanExt
|
||||
let span_id = span_context.span().span_context().span_id().to_string(); // Get the SpanId
|
||||
|
||||
// Audit events are logged
|
||||
let audit_entry = AuditEntry {
|
||||
version: "1.0".to_string(),
|
||||
event_type: "s3_put_object".to_string(),
|
||||
bucket,
|
||||
object,
|
||||
user,
|
||||
time: Utc::now().to_rfc3339(),
|
||||
user_agent: "rustfs.rs-client".to_string(),
|
||||
span_id,
|
||||
};
|
||||
audit_logger.log(audit_entry).await;
|
||||
|
||||
log_info("PUT operation completed");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
// #[cfg(feature = "audit-webhook")]
|
||||
// #[cfg(feature = "audit-kafka")]
|
||||
async fn test_main() {
|
||||
let telemetry = Telemetry::init();
|
||||
|
||||
// Initialize multiple audit objectives
|
||||
let audit_targets: Vec<Box<dyn AuditTarget>> = vec![
|
||||
Box::new(FileAuditTarget),
|
||||
// Box::new(KafkaAuditTarget::new("localhost:9092", "rustfs-audit")),
|
||||
// Box::new(WebhookAuditTarget::new("http://localhost:8080/audit")),
|
||||
];
|
||||
let audit_logger = AuditLogger::new(audit_targets);
|
||||
|
||||
// 创建根 Span 并执行操作
|
||||
// let tracer = global::tracer("main");
|
||||
// tracer.in_span("main_operation", |cx| {
|
||||
// Span::current().set_parent(cx);
|
||||
// log_info("Starting test async");
|
||||
// tokio::runtime::Runtime::new().unwrap().block_on(async {
|
||||
log_info("Starting test");
|
||||
// Test the PUT operation
|
||||
put_object(&audit_logger, "my-bucket".to_string(), "my-object.txt".to_string(), "user123".to_string()).await;
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
query_object(&audit_logger, "my-bucket".to_string(), "my-object.txt".to_string(), "user123".to_string()).await;
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
for i in 0..100 {
|
||||
put_object(
|
||||
&audit_logger,
|
||||
format!("my-bucket-{}", i),
|
||||
format!("my-object-{}", i),
|
||||
"user123".to_string(),
|
||||
)
|
||||
.await;
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
query_object(
|
||||
&audit_logger,
|
||||
format!("my-bucket-{}", i),
|
||||
format!("my-object-{}", i),
|
||||
"user123".to_string(),
|
||||
)
|
||||
.await;
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
// Wait for the export to complete
|
||||
tokio::time::sleep(Duration::from_secs(2)).await;
|
||||
log_info("Test completed");
|
||||
// });
|
||||
// });
|
||||
drop(telemetry); // Make sure to clean up
|
||||
}
|
||||
|
||||
#[instrument(fields(bucket, object, user))]
|
||||
async fn query_object(audit_logger: &AuditLogger, bucket: String, object: String, user: String) {
|
||||
let start_time = SystemTime::now();
|
||||
log_info("Starting query operation");
|
||||
|
||||
// Simulate the operation
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
// Record Metrics
|
||||
let meter = global::meter("rustfs.rs");
|
||||
let request_duration = meter.f64_histogram("s3_request_duration_seconds").build();
|
||||
request_duration.record(
|
||||
start_time.elapsed().unwrap().as_secs_f64(),
|
||||
&[opentelemetry::KeyValue::new("operation", "query_object")],
|
||||
);
|
||||
|
||||
// Gets the current span
|
||||
let span = Span::current();
|
||||
// Use 'OpenTelemetrySpanExt' to get 'SpanContext'
|
||||
let span_context = span.context(); // Get context via OpenTelemetrySpanExt
|
||||
let span_id = span_context.span().span_context().span_id().to_string(); // Get the SpanId
|
||||
query_one(user.clone());
|
||||
query_two(user.clone());
|
||||
query_three(user.clone());
|
||||
// Audit events are logged
|
||||
let audit_entry = AuditEntry {
|
||||
version: "1.0".to_string(),
|
||||
event_type: "s3_query_object".to_string(),
|
||||
bucket,
|
||||
object,
|
||||
user,
|
||||
time: Utc::now().to_rfc3339(),
|
||||
user_agent: "rustfs.rs-client".to_string(),
|
||||
span_id,
|
||||
};
|
||||
audit_logger.log(audit_entry).await;
|
||||
|
||||
log_info("query operation completed");
|
||||
}
|
||||
#[instrument(fields(user))]
|
||||
fn query_one(user: String) {
|
||||
// 初始化 OpenTelemetry Tracer
|
||||
let tracer = global::tracer("query_one");
|
||||
tracer.in_span("doing_work", |cx| {
|
||||
// Traced app logic here...
|
||||
Span::current().set_parent(cx);
|
||||
log_info("Doing work...");
|
||||
let current_span = Span::current();
|
||||
let span_context = current_span.context();
|
||||
let trace_id = span_context.clone().span().span_context().trace_id().to_string();
|
||||
let span_id = span_context.clone().span().span_context().span_id().to_string();
|
||||
log_info(format!("trace_id: {}, span_id: {}", trace_id, span_id).as_str());
|
||||
});
|
||||
log_info(format!("Starting query_one operation user:{}", user).as_str());
|
||||
}
|
||||
#[instrument(fields(user))]
|
||||
fn query_two(user: String) {
|
||||
log_info(format!("Starting query_two operation user:{}", user).as_str());
|
||||
}
|
||||
#[instrument(fields(user))]
|
||||
fn query_three(user: String) {
|
||||
log_info(format!("Starting query_three operation user: {}", user).as_str());
|
||||
}
|
||||
/// Log module initialization function
|
||||
///
|
||||
/// Return to Logger and Clean Guard
|
||||
pub fn init_logging(config: AppConfig) -> (Logger, telemetry::OtelGuard) {
|
||||
let guard = init_telemetry(&config.observability);
|
||||
let sinks = sink::create_sinks(&config);
|
||||
let logger = start_logger(&config, sinks);
|
||||
(logger, guard)
|
||||
}
|
||||
|
||||
+232
-40
@@ -1,46 +1,238 @@
|
||||
use tracing::{debug, error, info};
|
||||
use crate::{AppConfig, LogEntry, SerializableLevel, Sink};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc::{self, Receiver, Sender};
|
||||
|
||||
/// Log an info message
|
||||
///
|
||||
/// # Arguments
|
||||
/// msg: &str - The message to log
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::log_info;
|
||||
///
|
||||
/// log_info("This is an info message");
|
||||
/// ```
|
||||
pub fn log_info(msg: &str) {
|
||||
info!("{}", msg);
|
||||
/// Server log processor
|
||||
pub struct Logger {
|
||||
sender: Sender<LogEntry>, // Log sending channel
|
||||
queue_capacity: usize,
|
||||
}
|
||||
|
||||
/// Log an error message
|
||||
///
|
||||
/// # Arguments
|
||||
/// msg: &str - The message to log
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::log_error;
|
||||
///
|
||||
/// log_error("This is an error message");
|
||||
/// ```
|
||||
pub fn log_error(msg: &str) {
|
||||
error!("{}", msg);
|
||||
impl Logger {
|
||||
/// Create a new Logger instance
|
||||
/// Returns Logger and corresponding Receiver
|
||||
pub fn new(config: &AppConfig) -> (Self, Receiver<LogEntry>) {
|
||||
// Get queue capacity from configuration, or use default values 10000
|
||||
let queue_capacity = config.logger.queue_capacity.unwrap_or(10000);
|
||||
let (sender, receiver) = mpsc::channel(queue_capacity);
|
||||
(
|
||||
Logger {
|
||||
sender,
|
||||
queue_capacity,
|
||||
},
|
||||
receiver,
|
||||
)
|
||||
}
|
||||
|
||||
// Add a method to get queue capacity
|
||||
pub fn queue_capacity(&self) -> usize {
|
||||
self.queue_capacity
|
||||
}
|
||||
|
||||
/// Asynchronous logging of server logs
|
||||
/// Attach the log to the current Span and generate a separate Tracing Event
|
||||
#[tracing::instrument(skip(self), fields(log_source = "logger"))]
|
||||
pub async fn log(&self, entry: LogEntry) -> Result<(), LogError> {
|
||||
// Log messages to the current Span
|
||||
tracing::Span::current()
|
||||
.record("log_message", &entry.message)
|
||||
.record("source", &entry.source);
|
||||
|
||||
// Record queue utilization (if a certain threshold is exceeded)
|
||||
let queue_len = self.sender.capacity();
|
||||
let utilization = queue_len as f64 / self.queue_capacity as f64;
|
||||
if utilization > 0.8 {
|
||||
tracing::warn!("Log queue utilization high: {:.1}%", utilization * 100.0);
|
||||
}
|
||||
|
||||
// Generate independent Tracing Events with full LogEntry information
|
||||
// Generate corresponding events according to level
|
||||
match entry.level {
|
||||
SerializableLevel(tracing::Level::ERROR) => {
|
||||
tracing::error!(
|
||||
target: "server_logs",
|
||||
timestamp = %entry.timestamp,
|
||||
message = %entry.message,
|
||||
source = %entry.source,
|
||||
request_id = ?entry.request_id,
|
||||
user_id = ?entry.user_id,
|
||||
fields = ?entry.fields
|
||||
);
|
||||
}
|
||||
SerializableLevel(tracing::Level::WARN) => {
|
||||
tracing::warn!(
|
||||
target: "server_logs",
|
||||
timestamp = %entry.timestamp,
|
||||
message = %entry.message,
|
||||
source = %entry.source,
|
||||
request_id = ?entry.request_id,
|
||||
user_id = ?entry.user_id,
|
||||
fields = ?entry.fields
|
||||
);
|
||||
}
|
||||
SerializableLevel(tracing::Level::INFO) => {
|
||||
tracing::info!(
|
||||
target: "server_logs",
|
||||
timestamp = %entry.timestamp,
|
||||
message = %entry.message,
|
||||
source = %entry.source,
|
||||
request_id = ?entry.request_id,
|
||||
user_id = ?entry.user_id,
|
||||
fields = ?entry.fields
|
||||
);
|
||||
}
|
||||
SerializableLevel(tracing::Level::DEBUG) => {
|
||||
tracing::debug!(
|
||||
target: "server_logs",
|
||||
timestamp = %entry.timestamp,
|
||||
message = %entry.message,
|
||||
source = %entry.source,
|
||||
request_id = ?entry.request_id,
|
||||
user_id = ?entry.user_id,
|
||||
fields = ?entry.fields
|
||||
);
|
||||
}
|
||||
SerializableLevel(tracing::Level::TRACE) => {
|
||||
tracing::trace!(
|
||||
target: "server_logs",
|
||||
timestamp = %entry.timestamp,
|
||||
message = %entry.message,
|
||||
source = %entry.source,
|
||||
request_id = ?entry.request_id,
|
||||
user_id = ?entry.user_id,
|
||||
fields = ?entry.fields
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Send logs to asynchronous queues to improve error handling
|
||||
match self.sender.try_send(entry) {
|
||||
Ok(_) => Ok(()),
|
||||
Err(mpsc::error::TrySendError::Full(entry)) => {
|
||||
// Processing strategy when queue is full
|
||||
tracing::warn!("Log queue full, applying backpressure");
|
||||
match tokio::time::timeout(
|
||||
std::time::Duration::from_millis(500),
|
||||
self.sender.send(entry),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Ok(_)) => Ok(()),
|
||||
Ok(Err(_)) => Err(LogError::SendFailed("Channel closed")),
|
||||
Err(_) => Err(LogError::Timeout("Queue backpressure timeout")),
|
||||
}
|
||||
}
|
||||
Err(mpsc::error::TrySendError::Closed(_)) => {
|
||||
Err(LogError::SendFailed("Logger channel closed"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Add convenient methods to simplify logging
|
||||
// Fix the info() method, replacing None with an empty vector instead of the Option type
|
||||
pub async fn info(&self, message: &str, source: &str) -> Result<(), LogError> {
|
||||
self.log(LogEntry::new(
|
||||
tracing::Level::INFO,
|
||||
message.to_string(),
|
||||
source.to_string(),
|
||||
None,
|
||||
None,
|
||||
Vec::new(), // 使用空向量代替 None
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
/// Add warn() method
|
||||
pub async fn error(&self, message: &str, source: &str) -> Result<(), LogError> {
|
||||
self.log(LogEntry::new(
|
||||
tracing::Level::ERROR,
|
||||
message.to_string(),
|
||||
source.to_string(),
|
||||
None,
|
||||
None,
|
||||
Vec::new(),
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
/// Add warn() method
|
||||
pub async fn warn(&self, message: &str, source: &str) -> Result<(), LogError> {
|
||||
self.log(LogEntry::new(
|
||||
tracing::Level::WARN,
|
||||
message.to_string(),
|
||||
source.to_string(),
|
||||
None,
|
||||
None,
|
||||
Vec::new(),
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
/// Add debug() method
|
||||
pub async fn debug(&self, message: &str, source: &str) -> Result<(), LogError> {
|
||||
self.log(LogEntry::new(
|
||||
tracing::Level::DEBUG,
|
||||
message.to_string(),
|
||||
source.to_string(),
|
||||
None,
|
||||
None,
|
||||
Vec::new(),
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
/// Add trace() method
|
||||
pub async fn trace(&self, message: &str, source: &str) -> Result<(), LogError> {
|
||||
self.log(LogEntry::new(
|
||||
tracing::Level::TRACE,
|
||||
message.to_string(),
|
||||
source.to_string(),
|
||||
None,
|
||||
None,
|
||||
Vec::new(),
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
// Add extension methods with context information for more flexibility
|
||||
pub async fn info_with_context(
|
||||
&self,
|
||||
message: &str,
|
||||
source: &str,
|
||||
request_id: Option<String>,
|
||||
user_id: Option<String>,
|
||||
fields: Vec<(String, String)>,
|
||||
) -> Result<(), LogError> {
|
||||
self.log(LogEntry::new(
|
||||
tracing::Level::INFO,
|
||||
message.to_string(),
|
||||
source.to_string(),
|
||||
request_id,
|
||||
user_id,
|
||||
fields,
|
||||
))
|
||||
.await
|
||||
}
|
||||
|
||||
// Add elegant closing method
|
||||
pub async fn shutdown(self) -> Result<(), LogError> {
|
||||
drop(self.sender); //Close the sending end so that the receiver knows that there is no new message
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Log a debug message
|
||||
///
|
||||
/// # Arguments
|
||||
/// msg: &str - The message to log
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::log_debug;
|
||||
///
|
||||
/// log_debug("This is a debug message");
|
||||
/// ```
|
||||
pub fn log_debug(msg: &str) {
|
||||
debug!("{}", msg);
|
||||
// Define custom error type
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum LogError {
|
||||
#[error("Failed to send log: {0}")]
|
||||
SendFailed(&'static str),
|
||||
#[error("Operation timed out: {0}")]
|
||||
Timeout(&'static str),
|
||||
}
|
||||
|
||||
/// Start the log module
|
||||
pub fn start_logger(config: &AppConfig, sinks: Vec<Arc<dyn Sink>>) -> Logger {
|
||||
let (logger, receiver) = Logger::new(config);
|
||||
tokio::spawn(crate::worker::start_worker(receiver, sinks));
|
||||
logger
|
||||
}
|
||||
|
||||
@@ -0,0 +1,450 @@
|
||||
use crate::{AppConfig, LogEntry};
|
||||
use async_trait::async_trait;
|
||||
use std::sync::Arc;
|
||||
use tokio::fs::OpenOptions;
|
||||
use tokio::io;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
/// Sink Trait definition, asynchronously write logs
|
||||
#[async_trait]
|
||||
pub trait Sink: Send + Sync {
|
||||
async fn write(&self, entry: &LogEntry);
|
||||
}
|
||||
|
||||
#[cfg(feature = "kafka")]
|
||||
/// Kafka Sink Implementation
|
||||
pub struct KafkaSink {
|
||||
producer: rdkafka::producer::FutureProducer,
|
||||
topic: String,
|
||||
batch_size: usize,
|
||||
batch_timeout_ms: u64,
|
||||
entries: Arc<tokio::sync::Mutex<Vec<LogEntry>>>,
|
||||
last_flush: Arc<std::sync::atomic::AtomicU64>,
|
||||
}
|
||||
|
||||
#[cfg(feature = "kafka")]
|
||||
impl KafkaSink {
|
||||
/// Create a new KafkaSink instance
|
||||
pub fn new(producer: rdkafka::producer::FutureProducer, topic: String, batch_size: usize, batch_timeout_ms: u64) -> Self {
|
||||
// Create Arc-wrapped values first
|
||||
let entries = Arc::new(tokio::sync::Mutex::new(Vec::with_capacity(batch_size)));
|
||||
let last_flush = Arc::new(std::sync::atomic::AtomicU64::new(
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64,
|
||||
));
|
||||
let sink = KafkaSink {
|
||||
producer: producer.clone(),
|
||||
topic: topic.clone(),
|
||||
batch_size,
|
||||
batch_timeout_ms,
|
||||
entries: entries.clone(),
|
||||
last_flush: last_flush.clone(),
|
||||
};
|
||||
|
||||
// Start background flusher
|
||||
tokio::spawn(Self::periodic_flush(producer, topic, entries, last_flush, batch_timeout_ms));
|
||||
|
||||
sink
|
||||
}
|
||||
|
||||
/// Add a getter method to read the batch_timeout_ms field
|
||||
#[allow(dead_code)]
|
||||
pub fn batch_timeout(&self) -> u64 {
|
||||
self.batch_timeout_ms
|
||||
}
|
||||
|
||||
/// Add a method to dynamically adjust the timeout if needed
|
||||
#[allow(dead_code)]
|
||||
pub fn set_batch_timeout(&mut self, new_timeout_ms: u64) {
|
||||
self.batch_timeout_ms = new_timeout_ms;
|
||||
}
|
||||
|
||||
async fn periodic_flush(
|
||||
producer: rdkafka::producer::FutureProducer,
|
||||
topic: String,
|
||||
entries: Arc<tokio::sync::Mutex<Vec<LogEntry>>>,
|
||||
last_flush: Arc<std::sync::atomic::AtomicU64>,
|
||||
timeout_ms: u64,
|
||||
) {
|
||||
loop {
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(timeout_ms / 2)).await;
|
||||
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64;
|
||||
|
||||
let last = last_flush.load(std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
if now - last >= timeout_ms {
|
||||
let mut batch = entries.lock().await;
|
||||
if !batch.is_empty() {
|
||||
Self::send_batch(&producer, &topic, batch.drain(..).collect()).await;
|
||||
last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn send_batch(producer: &rdkafka::producer::FutureProducer, topic: &str, entries: Vec<LogEntry>) {
|
||||
for entry in entries {
|
||||
let payload = match serde_json::to_string(&entry) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
eprintln!("Failed to serialize log entry: {}", e);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let span_id = entry.timestamp.to_rfc3339();
|
||||
|
||||
let _ = producer
|
||||
.send(
|
||||
rdkafka::producer::FutureRecord::to(topic).payload(&payload).key(&span_id),
|
||||
std::time::Duration::from_secs(5),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "kafka")]
|
||||
#[async_trait]
|
||||
impl Sink for KafkaSink {
|
||||
async fn write(&self, entry: &LogEntry) {
|
||||
let mut batch = self.entries.lock().await;
|
||||
batch.push(entry.clone());
|
||||
|
||||
let should_flush_by_size = batch.len() >= self.batch_size;
|
||||
let should_flush_by_time = {
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64;
|
||||
let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
|
||||
now - last >= self.batch_timeout_ms
|
||||
};
|
||||
|
||||
if should_flush_by_size || should_flush_by_time {
|
||||
// Existing flush logic
|
||||
let entries_to_send: Vec<LogEntry> = batch.drain(..).collect();
|
||||
let producer = self.producer.clone();
|
||||
let topic = self.topic.clone();
|
||||
|
||||
self.last_flush.store(
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64,
|
||||
std::sync::atomic::Ordering::Relaxed,
|
||||
);
|
||||
|
||||
tokio::spawn(async move {
|
||||
KafkaSink::send_batch(&producer, &topic, entries_to_send).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "kafka")]
|
||||
impl Drop for KafkaSink {
|
||||
fn drop(&mut self) {
|
||||
// Perform any necessary cleanup here
|
||||
// For example, you might want to flush any remaining entries
|
||||
let producer = self.producer.clone();
|
||||
let topic = self.topic.clone();
|
||||
let entries = self.entries.clone();
|
||||
let last_flush = self.last_flush.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut batch = entries.lock().await;
|
||||
if !batch.is_empty() {
|
||||
KafkaSink::send_batch(&producer, &topic, batch.drain(..).collect()).await;
|
||||
last_flush.store(
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64,
|
||||
std::sync::atomic::Ordering::Relaxed,
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
eprintln!("Dropping KafkaSink with topic: {}", self.topic);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "webhook")]
|
||||
/// Webhook Sink Implementation
|
||||
pub struct WebhookSink {
|
||||
url: String,
|
||||
client: reqwest::Client,
|
||||
max_retries: usize,
|
||||
retry_delay_ms: u64,
|
||||
}
|
||||
|
||||
#[cfg(feature = "webhook")]
|
||||
impl WebhookSink {
|
||||
pub fn new(url: String, max_retries: usize, retry_delay_ms: u64) -> Self {
|
||||
WebhookSink {
|
||||
url,
|
||||
client: reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(10))
|
||||
.build()
|
||||
.unwrap_or_else(|_| reqwest::Client::new()),
|
||||
max_retries,
|
||||
retry_delay_ms,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "webhook")]
|
||||
#[async_trait]
|
||||
impl Sink for WebhookSink {
|
||||
async fn write(&self, entry: &LogEntry) {
|
||||
let mut retries = 0;
|
||||
let url = self.url.clone();
|
||||
let entry_clone = entry.clone();
|
||||
|
||||
while retries < self.max_retries {
|
||||
match self.client.post(&url).json(&entry_clone).send().await {
|
||||
Ok(response) if response.status().is_success() => {
|
||||
return;
|
||||
}
|
||||
_ => {
|
||||
retries += 1;
|
||||
if retries < self.max_retries {
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(
|
||||
self.retry_delay_ms * (1 << retries), // Exponential backoff
|
||||
))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
eprintln!("Failed to send log to webhook after {} retries", self.max_retries);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "webhook")]
|
||||
impl Drop for WebhookSink {
|
||||
fn drop(&mut self) {
|
||||
// Perform any necessary cleanup here
|
||||
// For example, you might want to log that the sink is being dropped
|
||||
eprintln!("Dropping WebhookSink with URL: {}", self.url);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "file")]
|
||||
/// File Sink Implementation
|
||||
pub struct FileSink {
|
||||
path: String,
|
||||
buffer_size: usize,
|
||||
writer: Arc<tokio::sync::Mutex<io::BufWriter<tokio::fs::File>>>,
|
||||
entry_count: std::sync::atomic::AtomicUsize,
|
||||
last_flush: std::sync::atomic::AtomicU64,
|
||||
flush_interval_ms: u64, // Time between flushes
|
||||
flush_threshold: usize, // Number of entries before flush
|
||||
}
|
||||
|
||||
#[cfg(feature = "file")]
|
||||
impl FileSink {
|
||||
#[allow(dead_code)]
|
||||
pub async fn new(
|
||||
path: String,
|
||||
buffer_size: usize,
|
||||
flush_interval_ms: u64,
|
||||
flush_threshold: usize,
|
||||
) -> Result<Self, std::io::Error> {
|
||||
let file = OpenOptions::new().append(true).create(true).open(&path).await?;
|
||||
|
||||
let writer = tokio::io::BufWriter::with_capacity(buffer_size, file);
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64;
|
||||
Ok(FileSink {
|
||||
path,
|
||||
buffer_size,
|
||||
writer: Arc::new(tokio::sync::Mutex::new(writer)),
|
||||
entry_count: std::sync::atomic::AtomicUsize::new(0),
|
||||
last_flush: std::sync::atomic::AtomicU64::new(now),
|
||||
flush_interval_ms,
|
||||
flush_threshold,
|
||||
})
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
async fn initialize_writer(&mut self) -> io::Result<()> {
|
||||
let file = tokio::fs::File::create(&self.path).await?;
|
||||
|
||||
// Use buffer_size to create a buffer writer with a specified capacity
|
||||
let buf_writer = io::BufWriter::with_capacity(self.buffer_size, file);
|
||||
|
||||
// Replace the original writer with the new Mutex
|
||||
self.writer = Arc::new(tokio::sync::Mutex::new(buf_writer));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Get the current buffer size
|
||||
#[allow(dead_code)]
|
||||
pub fn buffer_size(&self) -> usize {
|
||||
self.buffer_size
|
||||
}
|
||||
|
||||
// How to dynamically adjust the buffer size
|
||||
#[allow(dead_code)]
|
||||
pub async fn set_buffer_size(&mut self, new_size: usize) -> io::Result<()> {
|
||||
if self.buffer_size != new_size {
|
||||
self.buffer_size = new_size;
|
||||
// Reinitialize the writer directly, without checking is_some()
|
||||
self.initialize_writer().await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Check if flushing is needed based on count or time
|
||||
fn should_flush(&self) -> bool {
|
||||
// Check entry count threshold
|
||||
if self.entry_count.load(std::sync::atomic::Ordering::Relaxed) >= self.flush_threshold {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Check time threshold
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64;
|
||||
|
||||
let last = self.last_flush.load(std::sync::atomic::Ordering::Relaxed);
|
||||
now - last >= self.flush_interval_ms
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "file")]
|
||||
#[async_trait]
|
||||
impl Sink for FileSink {
|
||||
async fn write(&self, entry: &LogEntry) {
|
||||
let line = format!("{:?}\n", entry);
|
||||
let mut writer = self.writer.lock().await;
|
||||
|
||||
if let Err(e) = writer.write_all(line.as_bytes()).await {
|
||||
eprintln!("Failed to write log to file {}: {}", self.path, e);
|
||||
return;
|
||||
}
|
||||
|
||||
// Only flush periodically to improve performance
|
||||
// Logic to determine when to flush could be added here
|
||||
// Increment the entry count
|
||||
self.entry_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
// Check if we should flush
|
||||
if self.should_flush() {
|
||||
if let Err(e) = writer.flush().await {
|
||||
eprintln!("Failed to flush log file {}: {}", self.path, e);
|
||||
return;
|
||||
}
|
||||
|
||||
// Reset counters
|
||||
self.entry_count.store(0, std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64;
|
||||
|
||||
self.last_flush.store(now, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "file")]
|
||||
impl Drop for FileSink {
|
||||
fn drop(&mut self) {
|
||||
let writer = self.writer.clone();
|
||||
let path = self.path.clone();
|
||||
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
let mut writer = writer.lock().await;
|
||||
if let Err(e) = writer.flush().await {
|
||||
eprintln!("Failed to flush log file {}: {}", path, e);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a list of Sink instances
|
||||
pub fn create_sinks(config: &AppConfig) -> Vec<Arc<dyn Sink>> {
|
||||
let mut sinks: Vec<Arc<dyn Sink>> = Vec::new();
|
||||
|
||||
#[cfg(feature = "kafka")]
|
||||
if config.sinks.kafka.enabled {
|
||||
match rdkafka::config::ClientConfig::new()
|
||||
.set("bootstrap.servers", &config.sinks.kafka.bootstrap_servers)
|
||||
.set("message.timeout.ms", "5000")
|
||||
.create()
|
||||
{
|
||||
Ok(producer) => {
|
||||
sinks.push(Arc::new(KafkaSink::new(
|
||||
producer,
|
||||
config.sinks.kafka.topic.clone(),
|
||||
config.sinks.kafka.batch_size.unwrap_or(100),
|
||||
config.sinks.kafka.batch_timeout_ms.unwrap_or(1000),
|
||||
)));
|
||||
}
|
||||
Err(e) => eprintln!("Failed to create Kafka producer: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "webhook")]
|
||||
if config.sinks.webhook.enabled {
|
||||
sinks.push(Arc::new(WebhookSink::new(
|
||||
config.sinks.webhook.url.clone(),
|
||||
config.sinks.webhook.max_retries.unwrap_or(3),
|
||||
config.sinks.webhook.retry_delay_ms.unwrap_or(100),
|
||||
)));
|
||||
}
|
||||
|
||||
#[cfg(feature = "file")]
|
||||
{
|
||||
let path = if config.sinks.file.enabled {
|
||||
config.sinks.file.path.clone()
|
||||
} else {
|
||||
"default.log".to_string()
|
||||
};
|
||||
|
||||
// Use synchronous file operations
|
||||
let file_result = std::fs::OpenOptions::new().append(true).create(true).open(&path);
|
||||
|
||||
match file_result {
|
||||
Ok(file) => {
|
||||
let buffer_size = config.sinks.file.buffer_size.unwrap_or(8192);
|
||||
let writer = tokio::io::BufWriter::with_capacity(buffer_size, tokio::fs::File::from_std(file));
|
||||
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as u64;
|
||||
|
||||
sinks.push(Arc::new(FileSink {
|
||||
path: path.clone(),
|
||||
buffer_size,
|
||||
writer: Arc::new(tokio::sync::Mutex::new(writer)),
|
||||
entry_count: std::sync::atomic::AtomicUsize::new(0),
|
||||
last_flush: std::sync::atomic::AtomicU64::new(now),
|
||||
flush_interval_ms: config.sinks.file.flush_interval_ms.unwrap_or(1000),
|
||||
flush_threshold: config.sinks.file.flush_threshold.unwrap_or(100),
|
||||
}));
|
||||
}
|
||||
Err(e) => eprintln!("Failed to create file sink: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
sinks
|
||||
}
|
||||
+176
-103
@@ -1,149 +1,222 @@
|
||||
use crate::{get_local_ip_with_default, OtelConfig};
|
||||
use opentelemetry::trace::TracerProvider;
|
||||
use opentelemetry::{global, KeyValue};
|
||||
use opentelemetry_appender_tracing::layer;
|
||||
use opentelemetry_otlp::{self, WithExportConfig};
|
||||
use opentelemetry_otlp::WithExportConfig;
|
||||
use opentelemetry_sdk::logs::SdkLoggerProvider;
|
||||
use opentelemetry_sdk::{
|
||||
metrics::{MeterProviderBuilder, PeriodicReader, SdkMeterProvider},
|
||||
trace::{RandomIdGenerator, Sampler, SdkTracerProvider},
|
||||
Resource,
|
||||
};
|
||||
use opentelemetry_semantic_conventions::attribute::NETWORK_LOCAL_ADDRESS;
|
||||
use opentelemetry_semantic_conventions::{
|
||||
attribute::{DEPLOYMENT_ENVIRONMENT_NAME, SERVICE_NAME, SERVICE_VERSION},
|
||||
attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_NAME, SERVICE_VERSION},
|
||||
SCHEMA_URL,
|
||||
};
|
||||
use std::time::Duration;
|
||||
use tracing::{info, Level};
|
||||
use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
|
||||
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer};
|
||||
|
||||
/// Telemetry is a wrapper around the OpenTelemetry SDK Tracer and Meter providers.
|
||||
/// It initializes the global Tracer and Meter providers, and sets up the tracing subscriber.
|
||||
/// The Tracer and Meter providers are shut down when the Telemetry instance is dropped.
|
||||
/// This is a convenience struct to ensure that the global providers are properly initialized and shut down.
|
||||
/// A guard object that manages the lifecycle of OpenTelemetry components.
|
||||
///
|
||||
/// This struct holds references to the created OpenTelemetry providers and ensures
|
||||
/// they are properly shut down when the guard is dropped. It implements the RAII
|
||||
/// (Resource Acquisition Is Initialization) pattern for managing telemetry resources.
|
||||
///
|
||||
/// When this guard goes out of scope, it will automatically shut down:
|
||||
/// - The tracer provider (for distributed tracing)
|
||||
/// - The meter provider (for metrics collection)
|
||||
/// - The logger provider (for structured logging)
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// use rustfs_obs::Telemetry;
|
||||
///
|
||||
/// let _telemetry = Telemetry::init();
|
||||
/// ```
|
||||
pub struct Telemetry {
|
||||
/// use rustfs_obs::{init_telemetry, OtelConfig};
|
||||
///
|
||||
/// let config = OtelConfig::default();
|
||||
/// let otel_guard = init_telemetry(&config);
|
||||
///
|
||||
/// // The guard is kept alive for the duration of the application
|
||||
/// // When it's dropped, all telemetry components are properly shut down
|
||||
/// drop(otel_guard);
|
||||
/// ```
|
||||
pub struct OtelGuard {
|
||||
tracer_provider: SdkTracerProvider,
|
||||
meter_provider: SdkMeterProvider,
|
||||
logger_provider: SdkLoggerProvider,
|
||||
}
|
||||
|
||||
impl Telemetry {
|
||||
pub fn init() -> Self {
|
||||
// Define service resource information
|
||||
let resource = Resource::builder()
|
||||
.with_service_name("rustfs-service")
|
||||
.with_schema_url(
|
||||
[
|
||||
KeyValue::new(SERVICE_NAME, "rustfs-service"),
|
||||
KeyValue::new(SERVICE_VERSION, "0.1.0"),
|
||||
KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, "develop"),
|
||||
KeyValue::new(NETWORK_LOCAL_ADDRESS, "127.0.0.1"),
|
||||
],
|
||||
SCHEMA_URL,
|
||||
)
|
||||
.build();
|
||||
impl Drop for OtelGuard {
|
||||
fn drop(&mut self) {
|
||||
if let Err(err) = self.tracer_provider.shutdown() {
|
||||
eprintln!("Tracer shutdown error: {:?}", err);
|
||||
}
|
||||
if let Err(err) = self.meter_provider.shutdown() {
|
||||
eprintln!("Meter shutdown error: {:?}", err);
|
||||
}
|
||||
if let Err(err) = self.logger_provider.shutdown() {
|
||||
eprintln!("Logger shutdown error: {:?}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let tracer_exporter = opentelemetry_otlp::SpanExporter::builder()
|
||||
/// create OpenTelemetry Resource
|
||||
fn resource(config: &OtelConfig) -> Resource {
|
||||
Resource::builder()
|
||||
.with_service_name(config.service_name.clone())
|
||||
.with_schema_url(
|
||||
[
|
||||
KeyValue::new(SERVICE_NAME, config.service_name.clone()),
|
||||
KeyValue::new(SERVICE_VERSION, config.service_version.clone()),
|
||||
KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, config.deployment_environment.clone()),
|
||||
KeyValue::new(NETWORK_LOCAL_ADDRESS, get_local_ip_with_default()),
|
||||
],
|
||||
SCHEMA_URL,
|
||||
)
|
||||
.build()
|
||||
}
|
||||
|
||||
/// Initialize Meter Provider
|
||||
fn init_meter_provider(config: &OtelConfig) -> SdkMeterProvider {
|
||||
let mut builder = MeterProviderBuilder::default().with_resource(resource(config));
|
||||
// If endpoint is empty, use stdout output
|
||||
if config.endpoint.is_empty() {
|
||||
builder = builder.with_reader(
|
||||
PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default())
|
||||
.with_interval(std::time::Duration::from_secs(config.meter_interval))
|
||||
.build(),
|
||||
);
|
||||
} else {
|
||||
// If endpoint is not empty, use otlp output
|
||||
let exporter = opentelemetry_otlp::MetricExporter::builder()
|
||||
.with_tonic()
|
||||
.with_endpoint("http://localhost:4317")
|
||||
.with_protocol(opentelemetry_otlp::Protocol::Grpc)
|
||||
.with_timeout(Duration::from_secs(3))
|
||||
.build()
|
||||
.unwrap();
|
||||
|
||||
// Configure Tracer Provider
|
||||
let tracer_provider = SdkTracerProvider::builder()
|
||||
.with_sampler(Sampler::AlwaysOn)
|
||||
.with_id_generator(RandomIdGenerator::default())
|
||||
.with_resource(resource.clone())
|
||||
.with_batch_exporter(tracer_exporter)
|
||||
// .with_simple_exporter(opentelemetry_stdout::SpanExporter::default())
|
||||
.build();
|
||||
|
||||
let meter_exporter = opentelemetry_otlp::MetricExporter::builder()
|
||||
.with_tonic()
|
||||
.with_endpoint("http://localhost:4317")
|
||||
.with_protocol(opentelemetry_otlp::Protocol::Grpc)
|
||||
.with_timeout(Duration::from_secs(3))
|
||||
.with_endpoint(&config.endpoint)
|
||||
.with_temporality(opentelemetry_sdk::metrics::Temporality::default())
|
||||
.build()
|
||||
.unwrap();
|
||||
builder = builder.with_reader(
|
||||
PeriodicReader::builder(exporter)
|
||||
.with_interval(std::time::Duration::from_secs(config.meter_interval))
|
||||
.build(),
|
||||
);
|
||||
// If use_stdout is true, output to stdout at the same time
|
||||
if config.use_stdout {
|
||||
builder = builder.with_reader(
|
||||
PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default())
|
||||
.with_interval(std::time::Duration::from_secs(config.meter_interval))
|
||||
.build(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let meter_reader = PeriodicReader::builder(meter_exporter)
|
||||
.with_interval(Duration::from_secs(30))
|
||||
.build();
|
||||
let meter_provider = builder.build();
|
||||
global::set_meter_provider(meter_provider.clone());
|
||||
meter_provider
|
||||
}
|
||||
|
||||
// For debugging in development
|
||||
// let meter_stdout_reader = PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default()).build();
|
||||
/// Initialize Tracer Provider
|
||||
fn init_tracer_provider(config: &OtelConfig) -> SdkTracerProvider {
|
||||
let sampler = if config.sample_ratio > 0.0 && config.sample_ratio < 1.0 {
|
||||
Sampler::TraceIdRatioBased(config.sample_ratio)
|
||||
} else {
|
||||
Sampler::AlwaysOn
|
||||
};
|
||||
let builder = SdkTracerProvider::builder()
|
||||
.with_sampler(sampler)
|
||||
.with_id_generator(RandomIdGenerator::default())
|
||||
.with_resource(resource(config));
|
||||
|
||||
// Configure Meter Provider
|
||||
let meter_provider = MeterProviderBuilder::default()
|
||||
.with_resource(resource.clone())
|
||||
.with_reader(meter_reader)
|
||||
// .with_reader(meter_stdout_reader)
|
||||
.build();
|
||||
let tracer_provider = if config.endpoint.is_empty() {
|
||||
builder
|
||||
.with_simple_exporter(opentelemetry_stdout::SpanExporter::default())
|
||||
.build()
|
||||
} else {
|
||||
let exporter = opentelemetry_otlp::SpanExporter::builder()
|
||||
.with_tonic()
|
||||
.with_endpoint(&config.endpoint)
|
||||
.build()
|
||||
.unwrap();
|
||||
if config.use_stdout {
|
||||
builder
|
||||
.with_batch_exporter(exporter)
|
||||
.with_batch_exporter(opentelemetry_stdout::SpanExporter::default())
|
||||
} else {
|
||||
builder.with_batch_exporter(exporter)
|
||||
}
|
||||
.build()
|
||||
};
|
||||
|
||||
// Set global Tracer and Meter providers
|
||||
global::set_tracer_provider(tracer_provider.clone());
|
||||
global::set_meter_provider(meter_provider.clone());
|
||||
global::set_tracer_provider(tracer_provider.clone());
|
||||
tracer_provider
|
||||
}
|
||||
|
||||
let tracer = tracer_provider.tracer("rustfs-service");
|
||||
/// Initialize Telemetry
|
||||
pub fn init_telemetry(config: &OtelConfig) -> OtelGuard {
|
||||
let tracer_provider = init_tracer_provider(config);
|
||||
let meter_provider = init_meter_provider(config);
|
||||
let tracer = tracer_provider.tracer(config.service_name.clone());
|
||||
|
||||
// // let _stdout_exporter = opentelemetry_stdout::LogExporter::default();
|
||||
// let otlp_exporter = opentelemetry_otlp::LogExporter::builder()
|
||||
// .with_tonic()
|
||||
// .with_endpoint("http://localhost:4317")
|
||||
// // .with_timeout(Duration::from_secs(3))
|
||||
// // .with_protocol(opentelemetry_otlp::Protocol::Grpc)
|
||||
// .build()
|
||||
// .unwrap();
|
||||
// let provider: SdkLoggerProvider = SdkLoggerProvider::builder()
|
||||
// .with_resource(resource.clone())
|
||||
// .with_simple_exporter(otlp_exporter)
|
||||
// .build();
|
||||
// let filter_otel = EnvFilter::new("debug")
|
||||
// // .add_directive("hyper=off".parse().unwrap())
|
||||
// // .add_directive("opentelemetry=off".parse().unwrap())
|
||||
// // .add_directive("tonic=off".parse().unwrap())
|
||||
// // .add_directive("h2=off".parse().unwrap())
|
||||
// .add_directive("reqwest=off".parse().unwrap());
|
||||
// let otel_layer = layer::OpenTelemetryTracingBridge::new(&provider).with_filter(filter_otel);
|
||||
let logger_provider = if config.endpoint.is_empty() {
|
||||
SdkLoggerProvider::builder()
|
||||
.with_resource(resource(config))
|
||||
.with_simple_exporter(opentelemetry_stdout::LogExporter::default())
|
||||
.build()
|
||||
} else {
|
||||
let exporter = opentelemetry_otlp::LogExporter::builder()
|
||||
.with_tonic()
|
||||
.with_endpoint(&config.endpoint)
|
||||
.build()
|
||||
.unwrap();
|
||||
SdkLoggerProvider::builder()
|
||||
.with_resource(resource(config))
|
||||
.with_batch_exporter(exporter)
|
||||
.with_batch_exporter(opentelemetry_stdout::LogExporter::default())
|
||||
.build()
|
||||
};
|
||||
|
||||
let otel_layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
|
||||
// For the OpenTelemetry layer, add a tracing filter to filter events from
|
||||
// OpenTelemetry and its dependent crates (opentelemetry-otlp uses crates
|
||||
// like reqwest/tonic etc.) from being sent back to OTel itself, thus
|
||||
// preventing infinite telemetry generation. The filter levels are set as
|
||||
// follows:
|
||||
// - Allow `info` level and above by default.
|
||||
// - Restrict `opentelemetry`, `hyper`, `tonic`, and `reqwest` completely.
|
||||
// Note: This will also drop events from crates like `tonic` etc. even when
|
||||
// they are used outside the OTLP Exporter. For more details, see:
|
||||
// https://github.com/open-telemetry/opentelemetry-rust/issues/761
|
||||
let filter_otel = EnvFilter::new("info")
|
||||
.add_directive("hyper=off".parse().unwrap())
|
||||
.add_directive("opentelemetry=off".parse().unwrap())
|
||||
.add_directive("tonic=off".parse().unwrap())
|
||||
.add_directive("h2=off".parse().unwrap())
|
||||
.add_directive("reqwest=off".parse().unwrap());
|
||||
let otel_layer = otel_layer.with_filter(filter_otel);
|
||||
let registry = tracing_subscriber::registry()
|
||||
.with(tracing_subscriber::filter::LevelFilter::INFO)
|
||||
.with(OpenTelemetryLayer::new(tracer))
|
||||
.with(MetricsLayer::new(meter_provider.clone()))
|
||||
.with(otel_layer);
|
||||
if config.endpoint.is_empty() {
|
||||
// Create a new tracing::Fmt layer to print the logs to stdout. It has a
|
||||
// default filter of `info` level and above, and `debug` and above for logs
|
||||
// from OpenTelemetry crates. The filter levels can be customized as needed.
|
||||
let filter_fmt = EnvFilter::new("info").add_directive("opentelemetry=debug".parse().unwrap());
|
||||
let fmt_layer = tracing_subscriber::fmt::layer()
|
||||
.with_thread_names(true)
|
||||
.with_filter(filter_fmt);
|
||||
|
||||
// Configure `tracing subscriber`
|
||||
tracing_subscriber::registry()
|
||||
.with(tracing_subscriber::filter::LevelFilter::from_level(Level::DEBUG))
|
||||
registry
|
||||
.with(tracing_subscriber::fmt::layer().with_ansi(true))
|
||||
.with(MetricsLayer::new(meter_provider.clone()))
|
||||
.with(OpenTelemetryLayer::new(tracer))
|
||||
// .with(otel_layer)
|
||||
.with(fmt_layer)
|
||||
.init();
|
||||
info!(name: "my-event-name", target: "my-system", event_id = 20, user_name = "otel", user_email = "otel@opentelemetry.io", message = "This is an example message");
|
||||
Self {
|
||||
tracer_provider,
|
||||
meter_provider,
|
||||
}
|
||||
} else {
|
||||
registry.with(tracing_subscriber::fmt::layer().with_ansi(false)).init();
|
||||
println!("Logs and meter,tracer enabled");
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Telemetry {
|
||||
fn drop(&mut self) {
|
||||
if let Err(err) = self.tracer_provider.shutdown() {
|
||||
eprintln!("{err:?}");
|
||||
}
|
||||
if let Err(err) = self.meter_provider.shutdown() {
|
||||
eprintln!("{err:?}");
|
||||
}
|
||||
OtelGuard {
|
||||
tracer_provider,
|
||||
meter_provider,
|
||||
logger_provider,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
use local_ip_address::{local_ip, local_ipv6};
|
||||
use std::net::{IpAddr, Ipv4Addr};
|
||||
|
||||
/// Get the IP address of the machine
|
||||
///
|
||||
/// Priority is given to trying to get the IPv4 address, and if it fails, try to get the IPv6 address.
|
||||
/// If both fail to retrieve, None is returned.
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// * `Some(IpAddr)` - Native IP address (IPv4 or IPv6)
|
||||
/// * `None` - Unable to obtain any native IP address
|
||||
pub fn get_local_ip() -> Option<IpAddr> {
|
||||
local_ip().ok().or_else(|| local_ipv6().ok())
|
||||
}
|
||||
|
||||
/// Get the IP address of the machine as a string
|
||||
///
|
||||
/// If the IP address cannot be obtained, returns "127.0.0.1" as the default value.
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// * `String` - Native IP address (IPv4 or IPv6) as a string, or the default value
|
||||
pub fn get_local_ip_with_default() -> String {
|
||||
get_local_ip()
|
||||
.unwrap_or_else(|| IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))) // Provide a safe default value
|
||||
.to_string()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_get_local_ip() {
|
||||
match get_local_ip() {
|
||||
Some(ip) => println!("the ip address of this machine:{}", ip),
|
||||
None => println!("Unable to obtain the IP address of the machine"),
|
||||
}
|
||||
assert!(get_local_ip().is_some());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
use crate::{entry::LogEntry, sink::Sink};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc::Receiver;
|
||||
|
||||
/// Start the log processing worker thread
|
||||
pub async fn start_worker(receiver: Receiver<LogEntry>, sinks: Vec<Arc<dyn Sink>>) {
|
||||
let mut receiver = receiver;
|
||||
while let Some(entry) = receiver.recv().await {
|
||||
for sink in &sinks {
|
||||
sink.write(&entry).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user