Add pipeline_id tag to all pipeline-specific controller logs

crates/arroyo-controller/src/job_controller/leader_manager.rs

@@ -12,7 +12,7 @@ use arroyo_rpc::grpc::rpc::job_status_grpc_client::JobStatusGrpcClient;
 use arroyo_rpc::grpc::rpc::{JobState, JobStatusReq, JobStopMode, StopJobReq};
 use arroyo_rpc::identity::InjectWorkerId;
 use arroyo_rpc::{job_status_client, retry};
-use arroyo_types::{JobId, WorkerId};
+use arroyo_types::{JobId, PipelineId, WorkerId};
 use std::time::{Duration, Instant};
 use tonic::codegen::InterceptedService;
 use tonic::transport::Channel;
@@ -21,13 +21,15 @@ use tracing::{info, warn};
 pub struct LeaderManager {
     leader_client: JobStatusGrpcClient<InterceptedService<Channel, InjectWorkerId>>,
     pub job_id: JobId,
+    pub pipeline_id: PipelineId,
     pub generation: u64,
     pub last_heartbeat: Instant,
 }
 
 impl LeaderManager {
     pub async fn connect(
         job_id: JobId,
+        pipeline_id: PipelineId,
         generation: u64,
         worker_id: WorkerId,
         address: String,
@@ -43,11 +45,17 @@ impl LeaderManager {
             5,
             Duration::from_millis(100),
             Duration::from_secs(2),
-            |e| warn!(job_id = *job_id.0, message = "failed to connect to worker leader", error = ?e)
+            |e| warn!(
+                job_id = *job_id.0,
+                pipeline_id = *pipeline_id.0,
+                message = "failed to connect to worker leader",
+                error = ?e
+            )
         )?;
 
         Ok(Self {
             job_id,
+            pipeline_id,
             generation,
             leader_client,
             last_heartbeat: Instant::now(),
@@ -56,15 +64,23 @@ impl LeaderManager {
 
     pub async fn poll_leader_status(&mut self) -> anyhow::Result<rpc::JobStatus> {
         let response = retry!(
-            self.leader_client.get_job_status(JobStatusReq {
-                  job_id: self.job_id.to_string(),
-                   generation: self.generation,
-                }).await,
-                5,
-                Duration::from_millis(100),
-                Duration::from_secs(2),
-                |e| warn!(job_id = *self.job_id.0, message = "failed to poll for job status", error = ?e)
-        )?.into_inner();
+            self.leader_client
+                .get_job_status(JobStatusReq {
+                    job_id: self.job_id.to_string(),
+                    generation: self.generation,
+                })
+                .await,
+            5,
+            Duration::from_millis(100),
+            Duration::from_secs(2),
+            |e| warn!(
+                job_id = *self.job_id.0,
+                pipeline_id = *self.pipeline_id.0,
+                message = "failed to poll for job status",
+                error = ?e
+            )
+        )?
+        .into_inner();
 
         if response.job_id != *self.job_id.0 {
             bail!(
@@ -95,6 +111,7 @@ impl LeaderManager {
         info!(
             message = "sending stop request to leader",
             job_id = *self.job_id.0,
+            pipeline_id = *self.pipeline_id.0,
             stop_mode = ?stop_mode,
         );
 
@@ -190,7 +207,12 @@ where
                         }
                     }
                     Some(msg) => {
-                        warn!(job_id = *ctx.config.id, ?msg, "unexpected job message in leader leader mode");
+                        warn!(
+                            job_id = *ctx.config.id,
+                            pipeline_id = *ctx.pipeline_info.pipeline_id,
+                            ?msg,
+                            "unexpected job message in leader leader mode"
+                        );
                     }
                     None => {
                         panic!("job queue shut down");

crates/arroyo-controller/src/job_controller/mod.rs

@@ -183,14 +183,16 @@ impl JobController {
                     info!(
                         message = "setting new min epoch",
                         min_epoch = *min_epoch,
-                        job_id = *self.config.id
+                        job_id = *self.config.id,
+                        pipeline_id = *self.model.pipeline_id
                     );
                     self.model.min_epoch = min_epoch;
                 }
                 Ok(Err(e)) => {
                     error!(
                         message = "cleanup failed",
                         job_id = *self.config.id,
+                        pipeline_id = *self.model.pipeline_id,
                         error = format!("{:?}", e)
                     );
 
@@ -201,6 +203,7 @@ impl JobController {
                     error!(
                         message = "cleanup panicked",
                         job_id = *self.config.id,
+                        pipeline_id = *self.model.pipeline_id,
                         error = format!("{:?}", e)
                     );
 
@@ -311,7 +314,8 @@ impl JobController {
                 JobMessage::ConfigUpdate(c) if c.stop_mode == SqlStopMode::immediate => {
                     info!(
                         message = "stopping job immediately",
-                        job_id = *self.config.id
+                        job_id = *self.config.id,
+                        pipeline_id = *self.model.pipeline_id
                     );
                     self.stop_job(StopMode::Immediate).await?;
                 }
@@ -329,12 +333,14 @@ impl JobController {
     fn start_cleanup(&mut self, new_min: Epoch) -> JoinHandle<anyhow::Result<Epoch>> {
         let min_epoch = Epoch((*self.model.min_epoch).max(1));
         let job_id = self.config.id.clone();
+        let pipeline_id = self.model.pipeline_id.clone();
         let store = self.checkpoint_store.clone();
         let storage_role = self.model.storage_role.clone();
 
         info!(
             message = "Starting cleaning",
             job_id = *job_id,
+            pipeline_id = *pipeline_id,
             min_epoch = *min_epoch,
             new_min = *new_min
         );
@@ -371,6 +377,7 @@ impl JobController {
             info!(
                 message = "Finished cleaning",
                 job_id = *job_id,
+                pipeline_id = *pipeline_id,
                 min_epoch = *min_epoch,
                 new_min = *new_min,
                 duration = start.elapsed().as_secs_f32()

crates/arroyo-controller/src/lib.rs

@@ -195,16 +195,18 @@ impl ControllerGrpc for ControllerServer {
         &self,
         request: Request<RegisterWorkerReq>,
     ) -> Result<Response<RegisterWorkerResp>, Status> {
-        info!(
-            "Worker registered: {:?} -- {:?}",
-            request.get_ref(),
-            request.remote_addr()
-        );
-
+        let remote_addr = request.remote_addr();
         let req = request.into_inner();
         let worker = req
             .worker_context
             .ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
+        info!(
+            job_id = worker.job_id,
+            pipeline_id = worker.pipeline_id,
+            "Worker registered: {:?} -- {:?}",
+            worker,
+            remote_addr
+        );
 
         self.send_to_job_queue(
             &worker.job_id,
@@ -227,11 +229,17 @@ impl ControllerGrpc for ControllerServer {
         request: Request<TaskStartedReq>,
     ) -> Result<Response<TaskStartedResp>, Status> {
         let req = request.into_inner();
-        info!("task started: {:?}", req);
-
         let ctx = req
             .worker_context
             .ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
+        info!(
+            job_id = ctx.job_id,
+            pipeline_id = ctx.pipeline_id,
+            worker_id = ctx.worker_id,
+            task_id = req.task_id,
+            subtask_idx = req.subtask_idx,
+            "task started"
+        );
 
         self.send_to_job_queue(
             &ctx.job_id,
@@ -356,8 +364,12 @@ impl ControllerGrpc for ControllerServer {
             .worker_context
             .ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
         info!(
+            job_id = ctx.job_id,
+            pipeline_id = ctx.pipeline_id,
             "Worker {} initialization completed: success={}, error={:?}",
-            ctx.worker_id, req.success, req.error_message
+            ctx.worker_id,
+            req.success,
+            req.error_message
         );
 
         self.send_to_job_queue(
@@ -382,8 +394,17 @@ impl JobControllerGrpc for ControllerServer {
     ) -> Result<Response<TaskCheckpointEventResp>, Status> {
         let req = request.into_inner();
 
-        debug!("received task checkpoint event {:?}", req);
-        let job_id = job_id_from_context(&req.worker_context)?;
+        let ctx = req
+            .worker_context
+            .as_ref()
+            .ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
+        debug!(
+            job_id = ctx.job_id,
+            pipeline_id = ctx.pipeline_id,
+            "received task checkpoint event {:?}",
+            req
+        );
+        let job_id = ctx.job_id.clone();
 
         self.send_to_job_queue(
             &job_id,
@@ -400,8 +421,17 @@ impl JobControllerGrpc for ControllerServer {
     ) -> Result<Response<TaskCheckpointCompletedResp>, Status> {
         let req = request.into_inner();
 
-        debug!("received task checkpoint completed {:?}", req);
-        let job_id = job_id_from_context(&req.worker_context)?;
+        let ctx = req
+            .worker_context
+            .as_ref()
+            .ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
+        debug!(
+            job_id = ctx.job_id,
+            pipeline_id = ctx.pipeline_id,
+            "received task checkpoint completed {:?}",
+            req
+        );
+        let job_id = ctx.job_id.clone();
 
         self.send_to_job_queue(
             &job_id,
@@ -500,6 +530,7 @@ impl JobControllerGrpc for ControllerServer {
 
         info!(
             job_id = ctx.job_id,
+            pipeline_id = ctx.pipeline_id,
             operator_id = err.operator_id,
             message = "operator error",
             error_message = err.error,

crates/arroyo-controller/src/schedulers/embedded.rs

@@ -50,6 +50,8 @@ impl Scheduler for EmbeddedScheduler {
         let guard = shutdown.guard("embedded-worker");
 
         let job_id = req.job_id.clone();
+        let pipeline_id = req.pipeline_id.clone();
+        let log_job_id = job_id.clone();
         let generation = req.generation;
         let worker_id = WorkerId(self.worker_counter.fetch_add(1, Ordering::SeqCst));
         let handle = tokio::task::spawn(async move {
@@ -61,19 +63,38 @@ impl Scheduler for EmbeddedScheduler {
                 req.generation,
                 guard,
             );
+            let worker_job_id = log_job_id.clone();
+            let worker_pipeline_id = pipeline_id.clone();
 
             match tokio::task::spawn(async move {
                 if let Err(e) = server.start_async().await {
-                    error!("Failed to start worker {:?}: {:?}", worker_id, e);
+                    error!(
+                        job_id = *worker_job_id,
+                        pipeline_id = *worker_pipeline_id,
+                        "Failed to start worker {:?}: {:?}",
+                        worker_id,
+                        e
+                    );
                 }
             })
             .await
             {
                 Ok(_) => {
-                    info!("Worker {:?} finished", worker_id);
+                    info!(
+                        job_id = *log_job_id,
+                        pipeline_id = *pipeline_id,
+                        "Worker {:?} finished",
+                        worker_id
+                    );
                 }
                 Err(err) => {
-                    error!("Worker {:?} panicked: {:?}", worker_id, err);
+                    error!(
+                        job_id = *log_job_id,
+                        pipeline_id = *pipeline_id,
+                        "Worker {:?} panicked: {:?}",
+                        worker_id,
+                        err
+                    );
                 }
             }
         });

crates/arroyo-controller/src/schedulers/kubernetes/mod.rs

@@ -253,6 +253,7 @@ impl Scheduler for KubernetesScheduler {
 
         info!(
             job_id = *req.job_id,
+            pipeline_id = *req.pipeline_id,
             message = "starting workers on k8s",
             replicas = pods.len(),
             task_slots = req.slots
@@ -261,6 +262,7 @@ impl Scheduler for KubernetesScheduler {
         for pod in pods {
             info!(
                 job_id = *req.job_id,
+                pipeline_id = *req.pipeline_id,
                 message = "starting worker",
                 pod = pod.metadata.name
             );