Skip to content

Normalize all job_id logs #1116

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mwylde merged 2 commits into from
Jul 29, 2026
Merged

Normalize all job_id logs #1116

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
76ba8ae
Normalize all job_id logs
mwylde Jul 29, 2026
1785d4a
Upgrade kafka in CI
mwylde Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions crates/arroyo-controller/src/job_controller/leader_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ impl LeaderManager {
Duration::from_millis(100),
Duration::from_secs(2),
|e| warn!(
job_id = *job_id.0,
job_id = %job_id,
pipeline_id = *pipeline_id.0,
message = "failed to connect to worker leader",
error = ?e
Expand Down Expand Up @@ -74,7 +74,7 @@ impl LeaderManager {
Duration::from_millis(100),
Duration::from_secs(2),
|e| warn!(
job_id = *self.job_id.0,
job_id = %self.job_id,
pipeline_id = *self.pipeline_id.0,
message = "failed to poll for job status",
error = ?e
Expand Down Expand Up @@ -110,7 +110,7 @@ impl LeaderManager {
pub async fn stop_leader(&mut self, stop_mode: JobStopMode) -> anyhow::Result<()> {
info!(
message = "sending stop request to leader",
job_id = *self.job_id.0,
job_id = %self.job_id,
pipeline_id = *self.pipeline_id.0,
stop_mode = ?stop_mode,
);
Expand Down Expand Up @@ -208,7 +208,7 @@ where
}
Some(msg) => {
warn!(
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id,
?msg,
"unexpected job message in leader leader mode"
Expand Down
12 changes: 6 additions & 6 deletions crates/arroyo-controller/src/job_controller/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,15 +183,15 @@ 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,
job_id = %self.config.id,
pipeline_id = *self.model.pipeline_id,
error = format!("{:?}", e)
);
Expand All @@ -202,7 +202,7 @@ impl JobController {
Err(e) => {
error!(
message = "cleanup panicked",
job_id = *self.config.id,
job_id = %self.config.id,
pipeline_id = *self.model.pipeline_id,
error = format!("{:?}", e)
);
Expand Down Expand Up @@ -314,7 +314,7 @@ 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?;
Expand All @@ -339,7 +339,7 @@ impl JobController {

info!(
message = "Starting cleaning",
job_id = *job_id,
job_id = %job_id,
pipeline_id = *pipeline_id,
min_epoch = *min_epoch,
new_min = *new_min
Expand Down Expand Up @@ -376,7 +376,7 @@ impl JobController {

info!(
message = "Finished cleaning",
job_id = *job_id,
job_id = %job_id,
pipeline_id = *pipeline_id,
min_epoch = *min_epoch,
new_min = *new_min,
Expand Down
16 changes: 8 additions & 8 deletions crates/arroyo-controller/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,7 @@ impl ControllerGrpc for ControllerServer {
.worker_context
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
info!(
job_id = worker.job_id,
job_id = %worker.job_id,
pipeline_id = worker.pipeline_id,
"Worker registered: {:?} -- {:?}",
worker,
Expand Down Expand Up @@ -272,7 +272,7 @@ impl ControllerGrpc for ControllerServer {
.worker_context
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
info!(
job_id = ctx.job_id,
job_id = %ctx.job_id,
pipeline_id = ctx.pipeline_id,
worker_id = ctx.worker_id,
task_id = req.task_id,
Expand Down Expand Up @@ -403,7 +403,7 @@ impl ControllerGrpc for ControllerServer {
.worker_context
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
info!(
job_id = ctx.job_id,
job_id = %ctx.job_id,
pipeline_id = ctx.pipeline_id,
"Worker {} initialization completed: success={}, error={:?}",
ctx.worker_id,
Expand Down Expand Up @@ -438,7 +438,7 @@ impl JobControllerGrpc for ControllerServer {
.as_ref()
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
debug!(
job_id = ctx.job_id,
job_id = %ctx.job_id,
pipeline_id = ctx.pipeline_id,
"received task checkpoint event {:?}",
req
Expand All @@ -465,7 +465,7 @@ impl JobControllerGrpc for ControllerServer {
.as_ref()
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
debug!(
job_id = ctx.job_id,
job_id = %ctx.job_id,
pipeline_id = ctx.pipeline_id,
"received task checkpoint completed {:?}",
req
Expand Down Expand Up @@ -568,7 +568,7 @@ impl JobControllerGrpc for ControllerServer {
.ok_or_else(|| Status::invalid_argument("NonfatalErrorReq missing error"))?;

info!(
job_id = ctx.job_id,
job_id = %ctx.job_id,
pipeline_id = ctx.pipeline_id,
operator_id = err.operator_id,
message = "operator error",
Expand Down Expand Up @@ -662,7 +662,7 @@ impl ControllerServer {
Ok(())
}
} else {
warn!(message = "Received message for unknown job id", job_id);
warn!(message = "Received message for unknown job id", %job_id);
Err(Status::failed_precondition(format!(
"No job with id {job_id}"
)))
Expand Down Expand Up @@ -722,7 +722,7 @@ impl ControllerServer {

let state_context: StateContext =
serde_json::from_value(p.state_context.clone()).unwrap_or_else(|e| {
warn!(job_id = *id, original =? p.state_context, error =? e,
warn!(job_id = %id, original =? p.state_context, error =? e,
"failed to deserialize state context");
StateContext {
version: 1,
Expand Down
6 changes: 3 additions & 3 deletions crates/arroyo-controller/src/schedulers/embedded.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ impl Scheduler for EmbeddedScheduler {
match tokio::task::spawn(async move {
if let Err(e) = server.start_async().await {
error!(
job_id = *worker_job_id,
job_id = %worker_job_id,
pipeline_id = *worker_pipeline_id,
"Failed to start worker {:?}: {:?}",
worker_id,
Expand All @@ -81,15 +81,15 @@ impl Scheduler for EmbeddedScheduler {
{
Ok(_) => {
info!(
job_id = *log_job_id,
job_id = %log_job_id,
pipeline_id = *pipeline_id,
"Worker {:?} finished",
worker_id
);
}
Err(err) => {
error!(
job_id = *log_job_id,
job_id = %log_job_id,
pipeline_id = *pipeline_id,
"Worker {:?} panicked: {:?}",
worker_id,
Expand Down
4 changes: 2 additions & 2 deletions crates/arroyo-controller/src/schedulers/kubernetes/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,7 @@ impl Scheduler for KubernetesScheduler {
}

info!(
job_id = *req.job_id,
job_id = %req.job_id,
pipeline_id = *req.pipeline_id,
message = "starting workers on k8s",
replicas = pods.len(),
Expand All @@ -261,7 +261,7 @@ impl Scheduler for KubernetesScheduler {

for pod in pods {
info!(
job_id = *req.job_id,
job_id = %req.job_id,
pipeline_id = *req.pipeline_id,
message = "starting worker",
pod = pod.metadata.name
Expand Down
18 changes: 9 additions & 9 deletions crates/arroyo-controller/src/schedulers/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,7 @@ impl Scheduler for ProcessScheduler {
info!(
message = "Killing child",
worker_id,
job_id = *job_id,
job_id = %job_id,
pipeline_id = *pipeline_id
);
if let Err(e) = child.kill().await {
Expand Down Expand Up @@ -331,7 +331,7 @@ impl Scheduler for ProcessScheduler {
warn!(
message = "process scheduler worker exited without completion signal",
worker_id = worker_id.0,
job_id = *job_id,
job_id = %job_id,
pipeline_id = *pipeline_id,
);
}
Expand Down Expand Up @@ -627,7 +627,7 @@ impl NodeScheduler {
warn!(
message = "node not found for stop worker",
node_id = *worker.node_id.0,
job_id = *worker.job_id,
job_id = %worker.job_id,
pipeline_id = *worker.pipeline_id
);
return Ok(Some(worker_id));
Expand All @@ -639,7 +639,7 @@ impl NodeScheduler {

info!(
message = "stopping worker",
job_id = *worker.job_id,
job_id = %worker.job_id,
pipeline_id = *worker.pipeline_id,
node_id = *worker.node_id.0,
node_addr = node.addr,
Expand All @@ -648,7 +648,7 @@ impl NodeScheduler {

let Ok(mut client) = Self::client(&node).await else {
warn!(
job_id = *worker.job_id,
job_id = %worker.job_id,
pipeline_id = *worker.pipeline_id,
"Failed to connect to worker to stop; this likely means it is dead"
);
Expand All @@ -664,7 +664,7 @@ impl NodeScheduler {
.await
else {
warn!(
job_id = *worker.job_id,
job_id = %worker.job_id,
pipeline_id = *worker.pipeline_id,
"Failed to connect to worker to stop; this likely means it is dead"
);
Expand Down Expand Up @@ -732,14 +732,14 @@ impl Scheduler for NodeScheduler {
node.release_slots(worker_id, req.slots as usize);
} else {
warn!(
job_id,
%job_id,
pipeline_id, "Got worker finished message for unknown node {}", machine_id
);
}

if state.workers.remove(&worker_id).is_none() {
warn!(
job_id,
%job_id,
pipeline_id, "Got worker finished message for unknown worker {}", worker_id.0
);
}
Expand Down Expand Up @@ -803,7 +803,7 @@ impl Scheduler for NodeScheduler {

let slots_for_this_one = node.free_slots.min(to_schedule);
info!(
job_id = *start_pipeline_req.job_id,
job_id = %start_pipeline_req.job_id,
pipeline_id = *start_pipeline_req.pipeline_id,
"Scheduling {} slots on node {}",
slots_for_this_one,
Expand Down
2 changes: 1 addition & 1 deletion crates/arroyo-controller/src/states/checkpoint_stopping.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ impl State for CheckpointStopping {
match job_controller.checkpoint_finished().await {
Ok(done) => {
debug!(
job_id = *job_id,
job_id = %job_id,
pipeline_id = *pipeline_id,
"checked checkpoint, got {}, job_controller.finished(): {}, final_checkpoint_started: {}",
done,
Expand Down
2 changes: 1 addition & 1 deletion crates/arroyo-controller/src/states/failing.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ impl State for Failing {
warn!(
message = "failed to gracefully tear down cluster during failure",
error = format!("{:?}", e),
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id
);
}
Expand Down
4 changes: 2 additions & 2 deletions crates/arroyo-controller/src/states/leader_restarting.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ impl State for LeaderRestarting {
}
Some(msg) => {
warn!(
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id,
?msg,
"unexpected job message in leader mode"
Expand Down Expand Up @@ -91,7 +91,7 @@ impl State for LeaderRestarting {
}
RestartMode::force => {
info!(
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id,
"force restarting job, tearing down cluster"
);
Expand Down
6 changes: 3 additions & 3 deletions crates/arroyo-controller/src/states/leader_running.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ impl State for LeaderRunning {
}
Some(msg) => {
warn!(
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id,
msg =? msg,
"unexpected job message in leader mode"
Expand All @@ -157,7 +157,7 @@ impl State for LeaderRunning {
error!(
message = "Failed to update status",
error = format!("{:?}", e),
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id
);
ctx.status.restarts = restarts;
Expand Down Expand Up @@ -232,7 +232,7 @@ impl State for LeaderRunning {
warn!(
message = "error while polling leader status",
error = format!("{:?}", err),
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id
);
tokio::time::sleep(Duration::from_secs(2)).await;
Expand Down
6 changes: 3 additions & 3 deletions crates/arroyo-controller/src/states/leader_stopping.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ impl State for LeaderStopping {

info!(
msg = "waiting for workers to terminate",
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id
);
// TODO: we should watch the config queue and move immediately to force stop if requested
Expand All @@ -64,7 +64,7 @@ impl State for LeaderStopping {
Ok(Err(e)) => {
error!(
msg = "encountered error while waiting for job to stop gracefully; will try force-stopping",
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id,
error = e.to_string(),
);
Expand All @@ -79,7 +79,7 @@ impl State for LeaderStopping {
Err(_e) => {
error!(
msg = "timed out waiting for job to stop",
job_id = *ctx.config.id,
job_id = %ctx.config.id,
pipeline_id = *ctx.pipeline_info.pipeline_id,
);

Expand Down
Loading
Loading