feat: add line number and file info in log message (#709)

crates/arroyo-rpc/default.toml

@@ -82,3 +82,8 @@ user = "arroyo"
 password = "arroyo"
 
 [logging]
+format = "plaintext" # option: plaintext / json / logfmt 
+nonblocking = false
+enable-file-line = false
+enable-file-name = false 
+buffered-lines-limit = 4096

crates/arroyo-rpc/src/config.rs

@@ -629,6 +629,18 @@ pub struct LogConfig {
     /// Nonblocking logging may reduce tail latency at the cost of higher memory usage
     #[serde(default)]
     pub nonblocking: bool,
+
+    /// Set the number of lines to buffer before dropping logs or exerting backpressure on senders
+    /// Only valid when nonblocking is set to true
+    pub buffered_lines_limit: usize,
+
+    /// Set switch whether record file line number in log
+    #[serde(default)]
+    pub enable_file_line: bool,
+
+    /// Set switch whether record file name in log
+    #[serde(default)]
+    pub enable_file_name: bool,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize, Default)]

crates/arroyo-server-common/src/lib.rs

@@ -38,7 +38,7 @@ use tracing_subscriber::EnvFilter;
 use tracing_subscriber::Registry;
 
 use arroyo_rpc::config::{config, LogFormat};
-use tracing_appender::non_blocking::WorkerGuard;
+use tracing_appender::non_blocking::{NonBlockingBuilder, WorkerGuard};
 use tracing_log::LogTracer;
 use uuid::Uuid;
 
@@ -84,7 +84,9 @@ pub fn init_logging_with_filter(_name: &str, filter: EnvFilter) -> Option<Worker
         .add_directive("aws_config::profile::credentials=warn".parse().unwrap());
 
     let (nonblocking, guard) = if config().logging.nonblocking {
-        let (nonblocking, guard) = tracing_appender::non_blocking(std::io::stderr());
+        let (nonblocking, guard) = NonBlockingBuilder::default()
+            .buffered_lines_limit(config().logging.buffered_lines_limit)
+            .finish(std::io::stderr());
         (Some(nonblocking), Some(guard))
     } else {
         (None, None)
@@ -94,8 +96,8 @@ pub fn init_logging_with_filter(_name: &str, filter: EnvFilter) -> Option<Worker
         LogFormat::Plaintext => {
             register_log!(
                 tracing_subscriber::fmt::layer()
-                    .with_line_number(false)
-                    .with_file(false)
+                    .with_line_number(config().logging.enable_file_line)
+                    .with_file(config().logging.enable_file_name)
                     .with_span_events(FmtSpan::NONE),
                 nonblocking,
                 filter
@@ -104,6 +106,8 @@ pub fn init_logging_with_filter(_name: &str, filter: EnvFilter) -> Option<Worker
         LogFormat::Logfmt => {
             register_log!(
                 tracing_subscriber::fmt::layer()
+                    .with_line_number(config().logging.enable_file_line)
+                    .with_file(config().logging.enable_file_name)
                     .event_format(tracing_logfmt::EventsFormatter)
                     .fmt_fields(tracing_logfmt::FieldsFormatter),
                 nonblocking,
@@ -112,7 +116,10 @@ pub fn init_logging_with_filter(_name: &str, filter: EnvFilter) -> Option<Worker
         }
         LogFormat::Json => {
             register_log!(
-                tracing_subscriber::fmt::layer().event_format(Format::default().json()),
+                tracing_subscriber::fmt::layer()
+                    .with_line_number(config().logging.enable_file_line)
+                    .with_file(config().logging.enable_file_name)
+                    .event_format(Format::default().json()),
                 nonblocking,
                 filter
             )