#![allow(non_snake_case)]
use std::fmt::Debug;
use std::sync::{Arc, Mutex, Condvar};
use std::thread;
use std::time::Duration;
use std::collections::VecDeque;
use crate::core::RiResult;
use crate::fs::RiFileSystem;
use rand;
use serde_json::json;
use std::fs as stdfs;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
mod context;
pub use context::RiLogContext;
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass(eq, eq_int))]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RiLogLevel {
Debug,
Info,
Warn,
Error,
}
impl RiLogLevel {
pub fn as_str(&self) -> &'static str {
match self {
RiLogLevel::Debug => "DEBUG",
RiLogLevel::Info => "INFO",
RiLogLevel::Warn => "WARN",
RiLogLevel::Error => "ERROR",
}
}
pub fn color_block(&self) -> &'static str {
match self {
RiLogLevel::Debug => "🟦",
RiLogLevel::Info => "🟩",
RiLogLevel::Warn => "🟨",
RiLogLevel::Error => "🟥",
}
}
pub fn from_env() -> Option<Self> {
std::env::var("Ri_LOG_LEVEL").ok().and_then(|s| {
match s.to_ascii_uppercase().as_str() {
"DEBUG" => Some(RiLogLevel::Debug),
"INFO" => Some(RiLogLevel::Info),
"WARN" | "WARNING" => Some(RiLogLevel::Warn),
"ERROR" => Some(RiLogLevel::Error),
_ => None,
}
})
}
pub fn from_str(s: &str) -> Option<Self> {
match s.to_ascii_uppercase().as_str() {
"DEBUG" => Some(RiLogLevel::Debug),
"INFO" => Some(RiLogLevel::Info),
"WARN" | "WARNING" => Some(RiLogLevel::Warn),
"ERROR" => Some(RiLogLevel::Error),
_ => None,
}
}
}
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass(get_all, set_all))]
#[derive(Clone)]
pub struct RiLogConfig {
pub level: RiLogLevel,
pub console_enabled: bool,
pub file_enabled: bool,
pub sampling_default: f32,
pub file_name: String,
pub json_format: bool,
pub rotate_when: String,
pub max_bytes: u64,
pub color_blocks: bool,
}
#[cfg(feature = "pyo3")]
#[pyo3::prelude::pymethods]
impl RiLogConfig {
#[new]
fn py_new() -> Self {
Self::default()
}
#[staticmethod]
#[pyo3(name = "default")]
fn default_py() -> Self {
Self::default()
}
#[staticmethod]
#[pyo3(signature = (level="info", console_enabled=true, file_enabled=false, file_name="ri.log", json_format=false, max_bytes=10485760, color_blocks=true))]
fn create(
level: &str,
console_enabled: bool,
file_enabled: bool,
file_name: &str,
json_format: bool,
max_bytes: u64,
color_blocks: bool,
) -> Self {
let log_level = match level.to_uppercase().as_str() {
"DEBUG" => RiLogLevel::Debug,
"INFO" => RiLogLevel::Info,
"WARN" | "WARNING" => RiLogLevel::Warn,
"ERROR" => RiLogLevel::Error,
_ => RiLogLevel::Info,
};
Self {
level: log_level,
console_enabled,
file_enabled,
sampling_default: 1.0,
file_name: file_name.to_string(),
json_format,
rotate_when: "size".to_string(),
max_bytes,
color_blocks,
}
}
}
impl RiLogConfig {
pub fn from_config(config: &crate::config::RiConfig) -> Self {
let mut base = RiLogConfig::default();
if let Some(level_str) = config.get_str("log.level") {
if let Some(level) = RiLogLevel::from_str(level_str) {
base.level = level;
}
}
if let Some(v) = config.get_f32("log.sampling_default") {
base.sampling_default = v.clamp(0.0, 1.0);
}
if let Some(file_name) = config.get_str("log.file_name") {
if !file_name.is_empty() {
base.file_name = file_name.to_string();
}
}
if let Some(fmt) = config.get_str("log.file_format") {
if fmt.eq_ignore_ascii_case("json") {
base.json_format = true;
}
}
if let Some(rotate) = config.get_str("log.rotate_when") {
if !rotate.is_empty() {
base.rotate_when = rotate.to_string();
}
}
if let Some(v) = config.get_u64("log.max_bytes") {
if v > 0 {
base.max_bytes = v;
}
}
if let Some(v) = config.get_bool("log.console_enabled") {
base.console_enabled = v;
}
if let Some(v) = config.get_bool("log.file_enabled") {
base.file_enabled = v;
}
if let Some(v) = config.get_bool("log.color_blocks") {
base.color_blocks = v;
}
base
}
pub fn from_env() -> Self {
let mut base = RiLogConfig::default();
if let Some(level) = RiLogLevel::from_env() {
base.level = level;
}
if let Ok(v) = std::env::var("Ri_LOG_SAMPLING_DEFAULT") {
if let Ok(rate) = v.parse::<f32>() {
base.sampling_default = rate.clamp(0.0, 1.0);
}
}
if let Ok(file_name) = std::env::var("Ri_LOG_FILE_NAME") {
if !file_name.is_empty() {
base.file_name = file_name;
}
}
if let Ok(fmt) = std::env::var("Ri_LOG_FILE_FORMAT") {
if fmt.eq_ignore_ascii_case("json") {
base.json_format = true;
}
}
if let Ok(rotate) = std::env::var("Ri_LOG_ROTATE_WHEN") {
if !rotate.is_empty() {
base.rotate_when = rotate;
}
}
if let Ok(v) = std::env::var("Ri_LOG_MAX_BYTES") {
if let Ok(bytes) = v.parse::<u64>() {
if bytes > 0 {
base.max_bytes = bytes;
}
}
}
if let Ok(v) = std::env::var("Ri_LOG_CONSOLE_ENABLED") {
if v.eq_ignore_ascii_case("true") || v == "1" {
base.console_enabled = true;
} else if v.eq_ignore_ascii_case("false") || v == "0" {
base.console_enabled = false;
}
}
if let Ok(v) = std::env::var("Ri_LOG_FILE_ENABLED") {
if v.eq_ignore_ascii_case("true") || v == "1" {
base.file_enabled = true;
} else if v.eq_ignore_ascii_case("false") || v == "0" {
base.file_enabled = false;
}
}
if let Ok(v) = std::env::var("Ri_LOG_COLOR_BLOCKS") {
if v.eq_ignore_ascii_case("true") || v == "1" {
base.color_blocks = true;
} else if v.eq_ignore_ascii_case("false") || v == "0" {
base.color_blocks = false;
}
}
base
}
}
impl Default for RiLogConfig {
fn default() -> Self {
RiLogConfig {
level: RiLogLevel::Info,
console_enabled: true,
file_enabled: true,
sampling_default: 1.0,
file_name: "dms.log".to_string(),
json_format: false,
rotate_when: "size".to_string(),
max_bytes: 10 * 1024 * 1024,
color_blocks: true,
}
}
}
struct LogEntry {
level: RiLogLevel,
target: String,
message: String,
timestamp: String,
context: serde_json::Map<String, serde_json::Value>,
}
#[derive(Clone)]
struct LoggerImpl {
level: RiLogLevel,
#[allow(dead_code)]
fs: RiFileSystem,
sampling_default: f32,
#[allow(dead_code)]
console_enabled: bool,
#[allow(dead_code)]
file_enabled: bool,
#[allow(dead_code)]
file_name: String,
#[allow(dead_code)]
json_format: bool,
#[allow(dead_code)]
rotate_when: String,
#[allow(dead_code)]
max_bytes: u64,
#[allow(dead_code)]
color_blocks: bool,
log_cache: Arc<(Mutex<VecDeque<LogEntry>>, Condvar)>,
cache_size_limit: usize,
#[allow(dead_code)]
flush_interval_ms: u64,
#[allow(dead_code)]
shutdown_flag: Arc<Mutex<bool>>,
}
impl LoggerImpl {
fn new(config: &RiLogConfig, fs: RiFileSystem) -> Self {
let log_cache = Arc::new((Mutex::new(VecDeque::new()), Condvar::new()));
let shutdown_flag = Arc::new(Mutex::new(false));
let cache_size_limit = 1000;
let flush_interval_ms = 500;
let bg_log_cache = Arc::clone(&log_cache);
let bg_fs = fs.clone();
let bg_file_name = config.file_name.clone();
let bg_json_format = config.json_format;
let bg_rotate_when = config.rotate_when.clone();
let bg_max_bytes = config.max_bytes;
let bg_console_enabled = config.console_enabled;
let bg_color_blocks = config.color_blocks;
let bg_shutdown_flag = Arc::clone(&shutdown_flag);
thread::spawn(move || {
let mut last_flush = SystemTime::now();
loop {
let shutdown_flag = bg_shutdown_flag.lock().expect("Log shutdown flag lock poisoned");
if *shutdown_flag {
Self::flush_cache(
&bg_log_cache,
&bg_fs,
&bg_file_name,
bg_json_format,
&bg_rotate_when,
bg_max_bytes,
bg_console_enabled,
bg_color_blocks
).unwrap_or(());
break;
}
drop(shutdown_flag);
let now = SystemTime::now();
let time_since_last_flush = now.duration_since(last_flush).unwrap_or(Duration::from_millis(0));
let cache_len = bg_log_cache.0.lock().expect("Log cache lock poisoned").len();
if time_since_last_flush >= Duration::from_millis(flush_interval_ms) || cache_len >= cache_size_limit {
Self::flush_cache(
&bg_log_cache,
&bg_fs,
&bg_file_name,
bg_json_format,
&bg_rotate_when,
bg_max_bytes,
bg_console_enabled,
bg_color_blocks
).unwrap_or(());
last_flush = now;
}
let (lock, cvar) = &*bg_log_cache;
let _ = cvar.wait_timeout(lock.lock().unwrap(), Duration::from_millis(100)).unwrap();
}
});
LoggerImpl {
level: config.level,
fs,
sampling_default: config.sampling_default,
console_enabled: config.console_enabled,
file_enabled: config.file_enabled,
file_name: config.file_name.clone(),
json_format: config.json_format,
rotate_when: config.rotate_when.clone(),
max_bytes: config.max_bytes,
color_blocks: config.color_blocks,
log_cache,
cache_size_limit,
flush_interval_ms,
shutdown_flag,
}
}
fn flush_cache(
log_cache: &Arc<(Mutex<VecDeque<LogEntry>>, Condvar)>,
fs: &RiFileSystem,
file_name: &str,
json_format: bool,
rotate_when: &str,
max_bytes: u64,
console_enabled: bool,
color_blocks: bool
) -> RiResult<()> {
let (lock, _cvar) = &**log_cache;
let mut cache = lock.lock().unwrap();
if cache.is_empty() {
return Ok(());
}
let logs_to_flush: Vec<LogEntry> = cache.drain(..).collect();
drop(cache);
let mut file_logs = Vec::with_capacity(4);
let mut console_logs = Vec::with_capacity(4);
for log_entry in logs_to_flush {
let line = if json_format {
let mut log_obj = log_entry.context.clone();
if !log_obj.contains_key("level") {
log_obj.insert("level".to_string(), serde_json::Value::String(log_entry.level.as_str().to_string()));
}
if !log_obj.contains_key("target") {
log_obj.insert("target".to_string(), serde_json::Value::String(log_entry.target.clone()));
}
if !log_obj.contains_key("message") {
log_obj.insert("message".to_string(), serde_json::Value::String(log_entry.message.clone()));
}
if !log_obj.contains_key("timestamp") {
log_obj.insert("timestamp".to_string(), serde_json::Value::String(log_entry.timestamp.clone()));
}
serde_json::to_string(&log_obj)?
} else {
let ctx_kv: Vec<(String, String)> = log_entry.context.iter()
.filter(|(k, _)| *k != "timestamp" && *k != "level" && *k != "target" && *k != "message" && *k != "trace_id" && *k != "span_id")
.map(|(k, v)| (k.clone(), v.to_string()))
.collect();
let trace_info = match (log_entry.context.get("trace_id"), log_entry.context.get("span_id")) {
(Some(trace), Some(span)) => format!(" trace_id={trace} span_id={span}"),
(Some(trace), None) => format!(" trace_id={trace}"),
(None, Some(span)) => format!(" span_id={span}"),
(None, None) => String::new(),
};
let event = log_entry.context.get("event")
.and_then(|v| v.as_str())
.unwrap_or(&log_entry.target);
let ctx_display = if ctx_kv.is_empty() {
String::new()
} else {
let parts: Vec<String> = ctx_kv
.iter()
.map(|(k, v)| format!("{}={}", k, v))
.collect();
format!(" | {}", parts.join(", "))
};
let trace_display = if trace_info.is_empty() {
String::new()
} else {
format!(" | {}", trace_info.trim())
};
let color_block = if color_blocks {
log_entry.level.color_block()
} else {
""
};
let color_sep = if color_blocks { " | " } else { "" };
format!(
"{}{}{} | {:5} | {} | event={}{} | {}{}",
color_block,
color_sep,
log_entry.timestamp,
log_entry.level.as_str(),
log_entry.target,
event,
trace_display,
log_entry.message,
ctx_display,
)
};
if console_enabled {
console_logs.push(line.clone());
}
if !line.is_empty() {
file_logs.push(line);
}
}
if !console_logs.is_empty() {
for line in console_logs {
log::info!("{line}");
}
}
if !file_logs.is_empty() {
let log_file = fs.logs_dir().join(file_name);
if rotate_when.eq_ignore_ascii_case("size") && max_bytes > 0 {
if let Ok(meta) = stdfs::metadata(&log_file) {
if meta.len() >= max_bytes {
if let Some(parent) = log_file.parent() {
let base = log_file.file_name().and_then(|s| s.to_str()).unwrap_or("dms.log");
let ts = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|e| crate::core::RiError::Other(format!("timestamp error: {e}")))?;
let rotated = parent.join(format!("{}.{}", base, ts.as_millis()));
let _ = stdfs::rename(&log_file, &rotated);
}
}
}
}
let content = file_logs.join("\n") + "\n";
fs.append_text(&log_file, &content)?;
}
Ok(())
}
fn should_log(&self, level: RiLogLevel) -> bool {
(level as u8) >= (self.level as u8)
}
fn should_log_event(&self, event: &str) -> bool {
if let Some(event_sampling_rate) = self.get_event_sampling_rate(event) {
if event_sampling_rate >= 1.0 {
return true;
} else if event_sampling_rate <= 0.0 {
return false;
} else {
let r = rand::random::<f32>();
return r < event_sampling_rate;
}
}
if self.sampling_default >= 1.0 {
true
} else if self.sampling_default <= 0.0 {
false
} else {
let r = rand::random::<f32>();
r < self.sampling_default
}
}
fn get_event_sampling_rate(&self, event: &str) -> Option<f32> {
match event {
"database_query" => Some(0.1), "api_request" => Some(0.5), "cache_hit" => Some(0.05), "cache_miss" => Some(1.0), "error" => Some(1.0), "warning" => Some(0.8), _ => None, }
}
fn now_timestamp() -> String {
match SystemTime::now().duration_since(UNIX_EPOCH) {
Ok(dur) => {
let secs = dur.as_secs();
let millis = dur.subsec_millis();
format!("{secs}.{millis:03}Z")
}
Err(_) => "0.000Z".to_string(),
}
}
fn sanitize_log_message(message: &str) -> String {
let mut result = String::with_capacity(message.len().min(4096));
for c in message.chars() {
match c {
' '..='~' => {
if c == '\\' {
result.push_str("\\\\");
} else {
result.push(c);
}
}
c if c.is_alphanumeric() => {
result.push(c);
}
'\n' => result.push_str("\\n"),
'\r' => result.push_str("\\r"),
'\t' => result.push_str("\\t"),
_ => continue,
}
}
if result.len() > 4096 {
result.truncate(4096);
result.push_str("...[truncated]");
}
result
}
fn log_message<T: Debug>(&self, level: RiLogLevel, target: &str, message: T) -> RiResult<()> {
if !self.should_log(level) {
return Ok(());
}
let event = target; if !self.should_log_event(event) {
return Ok(());
}
let ts = Self::now_timestamp();
let message_str = format!("{message:?}");
let sanitized_message = Self::sanitize_log_message(&message_str);
let sanitized_target = Self::sanitize_log_message(target);
let ctx_kv = RiLogContext::get_all();
let mut log_entry_context = serde_json::Map::new();
log_entry_context.insert("timestamp".to_string(), json!(ts));
log_entry_context.insert("level".to_string(), json!(level.as_str()));
log_entry_context.insert("target".to_string(), json!(sanitized_target));
log_entry_context.insert("event".to_string(), json!(sanitized_target));
log_entry_context.insert("message".to_string(), json!(sanitized_message));
if let Some(trace_id) = RiLogContext::get_trace_id() {
log_entry_context.insert("trace_id".to_string(), json!(Self::sanitize_log_message(&trace_id)));
}
if let Some(span_id) = RiLogContext::get_span_id() {
log_entry_context.insert("span_id".to_string(), json!(Self::sanitize_log_message(&span_id)));
}
if let Some(parent_span_id) = RiLogContext::get_parent_span_id() {
log_entry_context.insert("parent_span_id".to_string(), json!(Self::sanitize_log_message(&parent_span_id)));
}
if !ctx_kv.is_empty() {
for (k, v) in ctx_kv.iter() {
log_entry_context.insert(
Self::sanitize_log_message(k),
json!(Self::sanitize_log_message(v))
);
}
}
let log_entry = LogEntry {
level,
target: sanitized_target,
message: sanitized_message,
timestamp: ts,
context: log_entry_context,
};
let (lock, cvar) = &*self.log_cache;
let mut cache = lock.lock().unwrap();
cache.push_back(log_entry);
if cache.len() >= self.cache_size_limit {
cvar.notify_one();
}
Ok(())
}
}
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
#[derive(Clone)]
pub struct RiLogger {
inner: LoggerImpl,
}
impl RiLogger {
pub fn new(config: &RiLogConfig, fs: RiFileSystem) -> Self {
let inner = LoggerImpl::new(config, fs);
RiLogger { inner }
}
pub fn debug<T: Debug>(&self, target: &str, message: T) -> RiResult<()> {
self.inner.log_message(RiLogLevel::Debug, target, message)
}
pub fn info<T: Debug>(&self, target: &str, message: T) -> RiResult<()> {
self.inner.log_message(RiLogLevel::Info, target, message)
}
pub fn warn<T: Debug>(&self, target: &str, message: T) -> RiResult<()> {
self.inner.log_message(RiLogLevel::Warn, target, message)
}
pub fn error<T: Debug>(&self, target: &str, message: T) -> RiResult<()> {
self.inner.log_message(RiLogLevel::Error, target, message)
}
}
#[cfg(feature = "pyo3")]
#[pyo3::prelude::pymethods]
impl RiLogger {
#[new]
fn py_new(config: RiLogConfig, fs: RiFileSystem) -> Self {
Self::new(&config, fs)
}
#[pyo3(name = "debug")]
fn py_debug(&self, target: &str, message: &str) -> pyo3::PyResult<()> {
self.inner
.log_message(RiLogLevel::Debug, target, message)
.map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
}
#[pyo3(name = "info")]
fn py_info(&self, target: &str, message: &str) -> pyo3::PyResult<()> {
self.inner
.log_message(RiLogLevel::Info, target, message)
.map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
}
#[pyo3(name = "warn")]
fn py_warn(&self, target: &str, message: &str) -> pyo3::PyResult<()> {
self.inner
.log_message(RiLogLevel::Warn, target, message)
.map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
}
#[pyo3(name = "error")]
fn py_error(&self, target: &str, message: &str) -> pyo3::PyResult<()> {
self.inner
.log_message(RiLogLevel::Error, target, message)
.map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
}
}