Add a config for tracing log rolling policy for both scheduler and executor (#487)

Co-authored-by: yangzhong <[email protected]>

Author: yahoNanJing

ballista/core/src/config.rs

@@ -286,6 +286,30 @@ impl parse_arg::ParseArgFromStr for TaskSchedulingPolicy {
     }
 }
 
+// an enum used to configure the log rolling policy
+// needs to be visible to code generated by configure_me
+#[derive(Clone, ArgEnum, Copy, Debug, serde::Deserialize)]
+pub enum LogRotationPolicy {
+    Minutely,
+    Hourly,
+    Daily,
+    Never,
+}
+
+impl std::str::FromStr for LogRotationPolicy {
+    type Err = String;
+
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        ArgEnum::from_str(s, true)
+    }
+}
+
+impl parse_arg::ParseArgFromStr for LogRotationPolicy {
+    fn describe_type<W: fmt::Write>(mut writer: W) -> fmt::Result {
+        write!(writer, "The log rotation policy")
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

ballista/executor/executor_config_spec.toml

@@ -125,3 +125,9 @@ name = "log_level_setting"
 type = "String"
 doc = "special log level for sub mod. link: https://docs.rs/env_logger/latest/env_logger/#enabling-logging. For example we want whole level is INFO but datafusion mode is DEBUG"
 default = "std::string::String::from(\"INFO,datafusion=INFO\")"
+
+[[param]]
+name = "log_rotation_policy"
+type = "ballista_core::config::LogRotationPolicy"
+doc = "Tracing log rotation policy, possible values: minutely, hourly, daily, never. Default: daily"
+default = "ballista_core::config::LogRotationPolicy::Daily"

ballista/executor/src/main.rs

@@ -33,7 +33,7 @@ use tokio::signal;
 use tokio::{fs, time};
 use uuid::Uuid;
 
-use ballista_core::config::TaskSchedulingPolicy;
+use ballista_core::config::{LogRotationPolicy, TaskSchedulingPolicy};
 use ballista_core::error::BallistaError;
 use ballista_core::serde::protobuf::{
     executor_registration, scheduler_grpc_client::SchedulerGrpcClient,
@@ -93,13 +93,32 @@ async fn main() -> Result<()> {
     let grpc_port = opt.bind_grpc_port;
     let log_dir = opt.log_dir;
     let print_thread_info = opt.print_thread_info;
-    let scheduler_name = format!("executor_{}_{}", bind_host, port);
+    let log_file_name_prefix = format!(
+        "executor_{}_{}",
+        external_host
+            .clone()
+            .unwrap_or_else(|| "localhost".to_string()),
+        port
+    );
 
     let rust_log = env::var(EnvFilter::DEFAULT_ENV);
     let log_filter = EnvFilter::new(rust_log.unwrap_or(special_mod_log_level));
     // File layer
     if let Some(log_dir) = log_dir {
-        let log_file = tracing_appender::rolling::daily(log_dir, &scheduler_name);
+        let log_file = match opt.log_rotation_policy {
+            LogRotationPolicy::Minutely => {
+                tracing_appender::rolling::minutely(log_dir, &log_file_name_prefix)
+            }
+            LogRotationPolicy::Hourly => {
+                tracing_appender::rolling::hourly(log_dir, &log_file_name_prefix)
+            }
+            LogRotationPolicy::Daily => {
+                tracing_appender::rolling::daily(log_dir, &log_file_name_prefix)
+            }
+            LogRotationPolicy::Never => {
+                tracing_appender::rolling::never(log_dir, &log_file_name_prefix)
+            }
+        };
         tracing_subscriber::fmt()
             .with_ansi(true)
             .with_thread_names(print_thread_info)

ballista/scheduler/scheduler_config_spec.toml

@@ -129,3 +129,9 @@ name = "log_level_setting"
 type = "String"
 doc = "special log level for sub mod. link: https://docs.rs/env_logger/latest/env_logger/#enabling-logging. For example we want whole level is INFO but datafusion mode is DEBUG"
 default = "std::string::String::from(\"INFO,datafusion=INFO\")"
+
+[[param]]
+name = "log_rotation_policy"
+type = "ballista_core::config::LogRotationPolicy"
+doc = "Tracing log rotation policy, possible values: minutely, hourly, daily, never. Default: daily"
+default = "ballista_core::config::LogRotationPolicy::Daily"

ballista/scheduler/src/main.rs

@@ -62,6 +62,7 @@ mod config {
 
 use ballista_core::utils::create_grpc_server;
 
+use ballista_core::config::LogRotationPolicy;
 use ballista_scheduler::config::SchedulerConfig;
 #[cfg(feature = "flight-sql")]
 use ballista_scheduler::flight_sql::FlightSqlServiceImpl;
@@ -170,7 +171,20 @@ async fn main() -> Result<()> {
     let log_filter = EnvFilter::new(rust_log.unwrap_or(special_mod_log_level));
     // File layer
     if let Some(log_dir) = log_dir {
-        let log_file = tracing_appender::rolling::daily(log_dir, &log_file_name_prefix);
+        let log_file = match opt.log_rotation_policy {
+            LogRotationPolicy::Minutely => {
+                tracing_appender::rolling::minutely(log_dir, &log_file_name_prefix)
+            }
+            LogRotationPolicy::Hourly => {
+                tracing_appender::rolling::hourly(log_dir, &log_file_name_prefix)
+            }
+            LogRotationPolicy::Daily => {
+                tracing_appender::rolling::daily(log_dir, &log_file_name_prefix)
+            }
+            LogRotationPolicy::Never => {
+                tracing_appender::rolling::never(log_dir, &log_file_name_prefix)
+            }
+        };
         tracing_subscriber::fmt()
             .with_ansi(true)
             .with_thread_names(print_thread_info)