Don't include worker_id in the service name in worker log (#1110)

Author: mwylde

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

@@ -54,9 +54,10 @@ pub const VERSION: &str = "0.16.0-dev";
 
 static CLUSTER_ID: OnceCell<String> = OnceCell::new();
 
-pub fn init_logging(name: &str) -> Option<WorkerGuard> {
+pub fn init_logging(name: &str, tags: &[(&str, &str)]) -> Option<WorkerGuard> {
     init_logging_with_filter(
         name,
+        tags,
         EnvFilter::builder()
             .with_default_directive(LevelFilter::INFO.into())
             .from_env_lossy(),
@@ -80,7 +81,11 @@ macro_rules! register_log {
     }};
 }
 
-pub fn init_logging_with_filter(name: &str, filter: EnvFilter) -> Option<WorkerGuard> {
+pub fn init_logging_with_filter(
+    name: &str,
+    tags: &[(&str, &str)],
+    filter: EnvFilter,
+) -> Option<WorkerGuard> {
     init_event_logger(Arc::new(AnalyticsEventLogger::new()));
 
     if let Err(e) = LogTracer::init() {
@@ -144,6 +149,10 @@ pub fn init_logging_with_filter(name: &str, filter: EnvFilter) -> Option<WorkerG
                 layer.add_static_field("pipeline_id", serde_json::json!(pipeline_id));
             }
 
+            for (k, v) in tags {
+                layer.add_static_field(*k, json!(*v));
+            }
+
             // add static fields
             for (key, value) in &config().logging.static_fields {
                 layer.add_static_field(key, serde_json::json!(value));

crates/arroyo-worker/src/lib.rs

@@ -41,7 +41,7 @@ use tokio::select;
 use tokio::sync::mpsc::{Receiver, Sender, channel};
 use tokio_stream::wrappers::TcpListenerStream;
 use tonic::{Request, Response, Status};
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 use crate::job_controller::controller::WorkerJobController;
 use crate::utils::{MAX_TASK_ERROR_FIELD_BYTES, maybe_truncate, to_d2};
@@ -1349,7 +1349,7 @@ impl JobControllerGrpc for LeaderServer {
 
         self.validate_req(req.worker_context.as_ref())?;
 
-        debug!(
+        trace!(
             worker_id =? self.state.worker_context.worker_id,
             ?req,
             "received heartbeat",

crates/arroyo/src/main.rs

@@ -377,7 +377,7 @@ where
 }
 
 async fn migrate(wait: Option<u32>) -> anyhow::Result<()> {
-    let _guard = arroyo_server_common::init_logging("migrate");
+    let _guard = arroyo_server_common::init_logging("migrate", &[]);
 
     let mut client = if let Some(wait) = wait {
         info!("Waiting for database to be ready to run migrations");
@@ -443,7 +443,7 @@ async fn migrate(wait: Option<u32>) -> anyhow::Result<()> {
 }
 
 async fn start_control_plane(service: CPService) {
-    let _guard = arroyo_server_common::init_logging(service.name());
+    let _guard = arroyo_server_common::init_logging(service.name(), &[]);
 
     let config = config::config();
 
@@ -494,11 +494,8 @@ async fn start_worker() {
     let server =
         WorkerServer::from_config(shutdown.guard("worker")).expect("Could not start worker");
 
-    let _guard = arroyo_server_common::init_logging(&format!(
-        "worker-{}-{}",
-        server.id().0,
-        server.job_id()
-    ));
+    let _guard =
+        arroyo_server_common::init_logging("worker", &[("worker_id", &server.id().to_string())]);
 
     shutdown.spawn_task("admin", start_admin_server("worker"));
     let token = shutdown.token();
@@ -516,15 +513,15 @@ async fn start_node() {
     let shutdown = Shutdown::new("node", SignalBehavior::Handle);
     let id = arroyo_node::start_server(shutdown.guard("node")).await;
 
-    let _guard = arroyo_server_common::init_logging(&format!("node-{}", id.0,));
+    let _guard = arroyo_server_common::init_logging("node", &[("node_id", &id)]);
 
     shutdown.spawn_task("admin", start_admin_server("worker"));
 
     Shutdown::handle_shutdown(shutdown.wait_for_shutdown(Duration::from_secs(30)).await);
 }
 
 async fn visualize(query: Input, open: bool) {
-    let _guard = arroyo_server_common::init_logging("visualize");
+    let _guard = arroyo_server_common::init_logging("visualize", &[]);
 
     let query = std::io::read_to_string(query).expect("Failed to read query");
 

crates/arroyo/src/run.rs

@@ -400,6 +400,7 @@ pub(crate) fn add_client_auth_headers(mut client_builder: ClientBuilder) -> Clie
 pub async fn run(args: RunArgs) {
     let _guard = arroyo_server_common::init_logging_with_filter(
         "pipeline",
+        &[],
         if env::var("RUST_LOG").is_err() {
             unsafe { set_var("RUST_LOG", "WARN") };
             EnvFilter::builder()