Skip to content

Commit 76ba8ae

Browse files
mwyldemwylde
committed
Normalize all job_id logs
1 parent e78f7b5 commit 76ba8ae

16 files changed

Lines changed: 97 additions & 97 deletions

File tree

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ impl LeaderManager {
4646
Duration::from_millis(100),
4747
Duration::from_secs(2),
4848
|e| warn!(
49-
job_id = *job_id.0,
49+
job_id = %job_id,
5050
pipeline_id = *pipeline_id.0,
5151
message = "failed to connect to worker leader",
5252
error = ?e
@@ -74,7 +74,7 @@ impl LeaderManager {
7474
Duration::from_millis(100),
7575
Duration::from_secs(2),
7676
|e| warn!(
77-
job_id = *self.job_id.0,
77+
job_id = %self.job_id,
7878
pipeline_id = *self.pipeline_id.0,
7979
message = "failed to poll for job status",
8080
error = ?e
@@ -110,7 +110,7 @@ impl LeaderManager {
110110
pub async fn stop_leader(&mut self, stop_mode: JobStopMode) -> anyhow::Result<()> {
111111
info!(
112112
message = "sending stop request to leader",
113-
job_id = *self.job_id.0,
113+
job_id = %self.job_id,
114114
pipeline_id = *self.pipeline_id.0,
115115
stop_mode = ?stop_mode,
116116
);
@@ -208,7 +208,7 @@ where
208208
}
209209
Some(msg) => {
210210
warn!(
211-
job_id = *ctx.config.id,
211+
job_id = %ctx.config.id,
212212
pipeline_id = *ctx.pipeline_info.pipeline_id,
213213
?msg,
214214
"unexpected job message in leader leader mode"

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -183,15 +183,15 @@ impl JobController {
183183
info!(
184184
message = "setting new min epoch",
185185
min_epoch = *min_epoch,
186-
job_id = *self.config.id,
186+
job_id = %self.config.id,
187187
pipeline_id = *self.model.pipeline_id
188188
);
189189
self.model.min_epoch = min_epoch;
190190
}
191191
Ok(Err(e)) => {
192192
error!(
193193
message = "cleanup failed",
194-
job_id = *self.config.id,
194+
job_id = %self.config.id,
195195
pipeline_id = *self.model.pipeline_id,
196196
error = format!("{:?}", e)
197197
);
@@ -202,7 +202,7 @@ impl JobController {
202202
Err(e) => {
203203
error!(
204204
message = "cleanup panicked",
205-
job_id = *self.config.id,
205+
job_id = %self.config.id,
206206
pipeline_id = *self.model.pipeline_id,
207207
error = format!("{:?}", e)
208208
);
@@ -314,7 +314,7 @@ impl JobController {
314314
JobMessage::ConfigUpdate(c) if c.stop_mode == SqlStopMode::immediate => {
315315
info!(
316316
message = "stopping job immediately",
317-
job_id = *self.config.id,
317+
job_id = %self.config.id,
318318
pipeline_id = *self.model.pipeline_id
319319
);
320320
self.stop_job(StopMode::Immediate).await?;
@@ -339,7 +339,7 @@ impl JobController {
339339

340340
info!(
341341
message = "Starting cleaning",
342-
job_id = *job_id,
342+
job_id = %job_id,
343343
pipeline_id = *pipeline_id,
344344
min_epoch = *min_epoch,
345345
new_min = *new_min
@@ -376,7 +376,7 @@ impl JobController {
376376

377377
info!(
378378
message = "Finished cleaning",
379-
job_id = *job_id,
379+
job_id = %job_id,
380380
pipeline_id = *pipeline_id,
381381
min_epoch = *min_epoch,
382382
new_min = *new_min,

‎crates/arroyo-controller/src/lib.rs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -240,7 +240,7 @@ impl ControllerGrpc for ControllerServer {
240240
.worker_context
241241
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
242242
info!(
243-
job_id = worker.job_id,
243+
job_id = %worker.job_id,
244244
pipeline_id = worker.pipeline_id,
245245
"Worker registered: {:?} -- {:?}",
246246
worker,
@@ -272,7 +272,7 @@ impl ControllerGrpc for ControllerServer {
272272
.worker_context
273273
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
274274
info!(
275-
job_id = ctx.job_id,
275+
job_id = %ctx.job_id,
276276
pipeline_id = ctx.pipeline_id,
277277
worker_id = ctx.worker_id,
278278
task_id = req.task_id,
@@ -403,7 +403,7 @@ impl ControllerGrpc for ControllerServer {
403403
.worker_context
404404
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
405405
info!(
406-
job_id = ctx.job_id,
406+
job_id = %ctx.job_id,
407407
pipeline_id = ctx.pipeline_id,
408408
"Worker {} initialization completed: success={}, error={:?}",
409409
ctx.worker_id,
@@ -438,7 +438,7 @@ impl JobControllerGrpc for ControllerServer {
438438
.as_ref()
439439
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
440440
debug!(
441-
job_id = ctx.job_id,
441+
job_id = %ctx.job_id,
442442
pipeline_id = ctx.pipeline_id,
443443
"received task checkpoint event {:?}",
444444
req
@@ -465,7 +465,7 @@ impl JobControllerGrpc for ControllerServer {
465465
.as_ref()
466466
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
467467
debug!(
468-
job_id = ctx.job_id,
468+
job_id = %ctx.job_id,
469469
pipeline_id = ctx.pipeline_id,
470470
"received task checkpoint completed {:?}",
471471
req
@@ -568,7 +568,7 @@ impl JobControllerGrpc for ControllerServer {
568568
.ok_or_else(|| Status::invalid_argument("NonfatalErrorReq missing error"))?;
569569

570570
info!(
571-
job_id = ctx.job_id,
571+
job_id = %ctx.job_id,
572572
pipeline_id = ctx.pipeline_id,
573573
operator_id = err.operator_id,
574574
message = "operator error",
@@ -662,7 +662,7 @@ impl ControllerServer {
662662
Ok(())
663663
}
664664
} else {
665-
warn!(message = "Received message for unknown job id", job_id);
665+
warn!(message = "Received message for unknown job id", %job_id);
666666
Err(Status::failed_precondition(format!(
667667
"No job with id {job_id}"
668668
)))
@@ -722,7 +722,7 @@ impl ControllerServer {
722722

723723
let state_context: StateContext =
724724
serde_json::from_value(p.state_context.clone()).unwrap_or_else(|e| {
725-
warn!(job_id = *id, original =? p.state_context, error =? e,
725+
warn!(job_id = %id, original =? p.state_context, error =? e,
726726
"failed to deserialize state context");
727727
StateContext {
728728
version: 1,

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ impl Scheduler for EmbeddedScheduler {
6969
match tokio::task::spawn(async move {
7070
if let Err(e) = server.start_async().await {
7171
error!(
72-
job_id = *worker_job_id,
72+
job_id = %worker_job_id,
7373
pipeline_id = *worker_pipeline_id,
7474
"Failed to start worker {:?}: {:?}",
7575
worker_id,
@@ -81,15 +81,15 @@ impl Scheduler for EmbeddedScheduler {
8181
{
8282
Ok(_) => {
8383
info!(
84-
job_id = *log_job_id,
84+
job_id = %log_job_id,
8585
pipeline_id = *pipeline_id,
8686
"Worker {:?} finished",
8787
worker_id
8888
);
8989
}
9090
Err(err) => {
9191
error!(
92-
job_id = *log_job_id,
92+
job_id = %log_job_id,
9393
pipeline_id = *pipeline_id,
9494
"Worker {:?} panicked: {:?}",
9595
worker_id,

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -252,7 +252,7 @@ impl Scheduler for KubernetesScheduler {
252252
}
253253

254254
info!(
255-
job_id = *req.job_id,
255+
job_id = %req.job_id,
256256
pipeline_id = *req.pipeline_id,
257257
message = "starting workers on k8s",
258258
replicas = pods.len(),
@@ -261,7 +261,7 @@ impl Scheduler for KubernetesScheduler {
261261

262262
for pod in pods {
263263
info!(
264-
job_id = *req.job_id,
264+
job_id = %req.job_id,
265265
pipeline_id = *req.pipeline_id,
266266
message = "starting worker",
267267
pod = pod.metadata.name

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

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -233,7 +233,7 @@ impl Scheduler for ProcessScheduler {
233233
info!(
234234
message = "Killing child",
235235
worker_id,
236-
job_id = *job_id,
236+
job_id = %job_id,
237237
pipeline_id = *pipeline_id
238238
);
239239
if let Err(e) = child.kill().await {
@@ -331,7 +331,7 @@ impl Scheduler for ProcessScheduler {
331331
warn!(
332332
message = "process scheduler worker exited without completion signal",
333333
worker_id = worker_id.0,
334-
job_id = *job_id,
334+
job_id = %job_id,
335335
pipeline_id = *pipeline_id,
336336
);
337337
}
@@ -627,7 +627,7 @@ impl NodeScheduler {
627627
warn!(
628628
message = "node not found for stop worker",
629629
node_id = *worker.node_id.0,
630-
job_id = *worker.job_id,
630+
job_id = %worker.job_id,
631631
pipeline_id = *worker.pipeline_id
632632
);
633633
return Ok(Some(worker_id));
@@ -639,7 +639,7 @@ impl NodeScheduler {
639639

640640
info!(
641641
message = "stopping worker",
642-
job_id = *worker.job_id,
642+
job_id = %worker.job_id,
643643
pipeline_id = *worker.pipeline_id,
644644
node_id = *worker.node_id.0,
645645
node_addr = node.addr,
@@ -648,7 +648,7 @@ impl NodeScheduler {
648648

649649
let Ok(mut client) = Self::client(&node).await else {
650650
warn!(
651-
job_id = *worker.job_id,
651+
job_id = %worker.job_id,
652652
pipeline_id = *worker.pipeline_id,
653653
"Failed to connect to worker to stop; this likely means it is dead"
654654
);
@@ -664,7 +664,7 @@ impl NodeScheduler {
664664
.await
665665
else {
666666
warn!(
667-
job_id = *worker.job_id,
667+
job_id = %worker.job_id,
668668
pipeline_id = *worker.pipeline_id,
669669
"Failed to connect to worker to stop; this likely means it is dead"
670670
);
@@ -732,14 +732,14 @@ impl Scheduler for NodeScheduler {
732732
node.release_slots(worker_id, req.slots as usize);
733733
} else {
734734
warn!(
735-
job_id,
735+
%job_id,
736736
pipeline_id, "Got worker finished message for unknown node {}", machine_id
737737
);
738738
}
739739

740740
if state.workers.remove(&worker_id).is_none() {
741741
warn!(
742-
job_id,
742+
%job_id,
743743
pipeline_id, "Got worker finished message for unknown worker {}", worker_id.0
744744
);
745745
}
@@ -803,7 +803,7 @@ impl Scheduler for NodeScheduler {
803803

804804
let slots_for_this_one = node.free_slots.min(to_schedule);
805805
info!(
806-
job_id = *start_pipeline_req.job_id,
806+
job_id = %start_pipeline_req.job_id,
807807
pipeline_id = *start_pipeline_req.pipeline_id,
808808
"Scheduling {} slots on node {}",
809809
slots_for_this_one,

‎crates/arroyo-controller/src/states/checkpoint_stopping.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ impl State for CheckpointStopping {
2828
match job_controller.checkpoint_finished().await {
2929
Ok(done) => {
3030
debug!(
31-
job_id = *job_id,
31+
job_id = %job_id,
3232
pipeline_id = *pipeline_id,
3333
"checked checkpoint, got {}, job_controller.finished(): {}, final_checkpoint_started: {}",
3434
done,

‎crates/arroyo-controller/src/states/failing.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ impl State for Failing {
1818
warn!(
1919
message = "failed to gracefully tear down cluster during failure",
2020
error = format!("{:?}", e),
21-
job_id = *ctx.config.id,
21+
job_id = %ctx.config.id,
2222
pipeline_id = *ctx.pipeline_info.pipeline_id
2323
);
2424
}

‎crates/arroyo-controller/src/states/leader_restarting.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ impl State for LeaderRestarting {
5858
}
5959
Some(msg) => {
6060
warn!(
61-
job_id = *ctx.config.id,
61+
job_id = %ctx.config.id,
6262
pipeline_id = *ctx.pipeline_info.pipeline_id,
6363
?msg,
6464
"unexpected job message in leader mode"
@@ -91,7 +91,7 @@ impl State for LeaderRestarting {
9191
}
9292
RestartMode::force => {
9393
info!(
94-
job_id = *ctx.config.id,
94+
job_id = %ctx.config.id,
9595
pipeline_id = *ctx.pipeline_info.pipeline_id,
9696
"force restarting job, tearing down cluster"
9797
);

‎crates/arroyo-controller/src/states/leader_running.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ impl State for LeaderRunning {
138138
}
139139
Some(msg) => {
140140
warn!(
141-
job_id = *ctx.config.id,
141+
job_id = %ctx.config.id,
142142
pipeline_id = *ctx.pipeline_info.pipeline_id,
143143
msg =? msg,
144144
"unexpected job message in leader mode"
@@ -157,7 +157,7 @@ impl State for LeaderRunning {
157157
error!(
158158
message = "Failed to update status",
159159
error = format!("{:?}", e),
160-
job_id = *ctx.config.id,
160+
job_id = %ctx.config.id,
161161
pipeline_id = *ctx.pipeline_info.pipeline_id
162162
);
163163
ctx.status.restarts = restarts;
@@ -232,7 +232,7 @@ impl State for LeaderRunning {
232232
warn!(
233233
message = "error while polling leader status",
234234
error = format!("{:?}", err),
235-
job_id = *ctx.config.id,
235+
job_id = %ctx.config.id,
236236
pipeline_id = *ctx.pipeline_info.pipeline_id
237237
);
238238
tokio::time::sleep(Duration::from_secs(2)).await;

0 commit comments

Comments
 (0)