Skip to content

Commit 859ff6d

Browse files
mwyldemwylde
authored
Add pipeline_id tag to all pipeline-specific controller logs (ArroyoSystems#1113)
* Add pipeline_id tag to all pipeline controller logs * Upgrade json-yaml to address CVE-2026-59869
1 parent c4cb7d3 commit 859ff6d

17 files changed

Lines changed: 399 additions & 106 deletions

File tree

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

Lines changed: 34 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ use arroyo_rpc::grpc::rpc::job_status_grpc_client::JobStatusGrpcClient;
1212
use arroyo_rpc::grpc::rpc::{JobState, JobStatusReq, JobStopMode, StopJobReq};
1313
use arroyo_rpc::identity::InjectWorkerId;
1414
use arroyo_rpc::{job_status_client, retry};
15-
use arroyo_types::{JobId, WorkerId};
15+
use arroyo_types::{JobId, PipelineId, WorkerId};
1616
use std::time::{Duration, Instant};
1717
use tonic::codegen::InterceptedService;
1818
use tonic::transport::Channel;
@@ -21,13 +21,15 @@ use tracing::{info, warn};
2121
pub struct LeaderManager {
2222
leader_client: JobStatusGrpcClient<InterceptedService<Channel, InjectWorkerId>>,
2323
pub job_id: JobId,
24+
pub pipeline_id: PipelineId,
2425
pub generation: u64,
2526
pub last_heartbeat: Instant,
2627
}
2728

2829
impl LeaderManager {
2930
pub async fn connect(
3031
job_id: JobId,
32+
pipeline_id: PipelineId,
3133
generation: u64,
3234
worker_id: WorkerId,
3335
address: String,
@@ -43,11 +45,17 @@ impl LeaderManager {
4345
5,
4446
Duration::from_millis(100),
4547
Duration::from_secs(2),
46-
|e| warn!(job_id = *job_id.0, message = "failed to connect to worker leader", error = ?e)
48+
|e| warn!(
49+
job_id = *job_id.0,
50+
pipeline_id = *pipeline_id.0,
51+
message = "failed to connect to worker leader",
52+
error = ?e
53+
)
4754
)?;
4855

4956
Ok(Self {
5057
job_id,
58+
pipeline_id,
5159
generation,
5260
leader_client,
5361
last_heartbeat: Instant::now(),
@@ -56,15 +64,23 @@ impl LeaderManager {
5664

5765
pub async fn poll_leader_status(&mut self) -> anyhow::Result<rpc::JobStatus> {
5866
let response = retry!(
59-
self.leader_client.get_job_status(JobStatusReq {
60-
job_id: self.job_id.to_string(),
61-
generation: self.generation,
62-
}).await,
63-
5,
64-
Duration::from_millis(100),
65-
Duration::from_secs(2),
66-
|e| warn!(job_id = *self.job_id.0, message = "failed to poll for job status", error = ?e)
67-
)?.into_inner();
67+
self.leader_client
68+
.get_job_status(JobStatusReq {
69+
job_id: self.job_id.to_string(),
70+
generation: self.generation,
71+
})
72+
.await,
73+
5,
74+
Duration::from_millis(100),
75+
Duration::from_secs(2),
76+
|e| warn!(
77+
job_id = *self.job_id.0,
78+
pipeline_id = *self.pipeline_id.0,
79+
message = "failed to poll for job status",
80+
error = ?e
81+
)
82+
)?
83+
.into_inner();
6884

6985
if response.job_id != *self.job_id.0 {
7086
bail!(
@@ -95,6 +111,7 @@ impl LeaderManager {
95111
info!(
96112
message = "sending stop request to leader",
97113
job_id = *self.job_id.0,
114+
pipeline_id = *self.pipeline_id.0,
98115
stop_mode = ?stop_mode,
99116
);
100117

@@ -190,7 +207,12 @@ where
190207
}
191208
}
192209
Some(msg) => {
193-
warn!(job_id = *ctx.config.id, ?msg, "unexpected job message in leader leader mode");
210+
warn!(
211+
job_id = *ctx.config.id,
212+
pipeline_id = *ctx.pipeline_info.pipeline_id,
213+
?msg,
214+
"unexpected job message in leader leader mode"
215+
);
194216
}
195217
None => {
196218
panic!("job queue shut down");

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

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -183,14 +183,16 @@ 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,
187+
pipeline_id = *self.model.pipeline_id
187188
);
188189
self.model.min_epoch = min_epoch;
189190
}
190191
Ok(Err(e)) => {
191192
error!(
192193
message = "cleanup failed",
193194
job_id = *self.config.id,
195+
pipeline_id = *self.model.pipeline_id,
194196
error = format!("{:?}", e)
195197
);
196198

@@ -201,6 +203,7 @@ impl JobController {
201203
error!(
202204
message = "cleanup panicked",
203205
job_id = *self.config.id,
206+
pipeline_id = *self.model.pipeline_id,
204207
error = format!("{:?}", e)
205208
);
206209

@@ -311,7 +314,8 @@ impl JobController {
311314
JobMessage::ConfigUpdate(c) if c.stop_mode == SqlStopMode::immediate => {
312315
info!(
313316
message = "stopping job immediately",
314-
job_id = *self.config.id
317+
job_id = *self.config.id,
318+
pipeline_id = *self.model.pipeline_id
315319
);
316320
self.stop_job(StopMode::Immediate).await?;
317321
}
@@ -329,12 +333,14 @@ impl JobController {
329333
fn start_cleanup(&mut self, new_min: Epoch) -> JoinHandle<anyhow::Result<Epoch>> {
330334
let min_epoch = Epoch((*self.model.min_epoch).max(1));
331335
let job_id = self.config.id.clone();
336+
let pipeline_id = self.model.pipeline_id.clone();
332337
let store = self.checkpoint_store.clone();
333338
let storage_role = self.model.storage_role.clone();
334339

335340
info!(
336341
message = "Starting cleaning",
337342
job_id = *job_id,
343+
pipeline_id = *pipeline_id,
338344
min_epoch = *min_epoch,
339345
new_min = *new_min
340346
);
@@ -371,6 +377,7 @@ impl JobController {
371377
info!(
372378
message = "Finished cleaning",
373379
job_id = *job_id,
380+
pipeline_id = *pipeline_id,
374381
min_epoch = *min_epoch,
375382
new_min = *new_min,
376383
duration = start.elapsed().as_secs_f32()

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

Lines changed: 44 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -195,16 +195,18 @@ impl ControllerGrpc for ControllerServer {
195195
&self,
196196
request: Request<RegisterWorkerReq>,
197197
) -> Result<Response<RegisterWorkerResp>, Status> {
198-
info!(
199-
"Worker registered: {:?} -- {:?}",
200-
request.get_ref(),
201-
request.remote_addr()
202-
);
203-
198+
let remote_addr = request.remote_addr();
204199
let req = request.into_inner();
205200
let worker = req
206201
.worker_context
207202
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
203+
info!(
204+
job_id = worker.job_id,
205+
pipeline_id = worker.pipeline_id,
206+
"Worker registered: {:?} -- {:?}",
207+
worker,
208+
remote_addr
209+
);
208210

209211
self.send_to_job_queue(
210212
&worker.job_id,
@@ -227,11 +229,17 @@ impl ControllerGrpc for ControllerServer {
227229
request: Request<TaskStartedReq>,
228230
) -> Result<Response<TaskStartedResp>, Status> {
229231
let req = request.into_inner();
230-
info!("task started: {:?}", req);
231-
232232
let ctx = req
233233
.worker_context
234234
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
235+
info!(
236+
job_id = ctx.job_id,
237+
pipeline_id = ctx.pipeline_id,
238+
worker_id = ctx.worker_id,
239+
task_id = req.task_id,
240+
subtask_idx = req.subtask_idx,
241+
"task started"
242+
);
235243

236244
self.send_to_job_queue(
237245
&ctx.job_id,
@@ -356,8 +364,12 @@ impl ControllerGrpc for ControllerServer {
356364
.worker_context
357365
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
358366
info!(
367+
job_id = ctx.job_id,
368+
pipeline_id = ctx.pipeline_id,
359369
"Worker {} initialization completed: success={}, error={:?}",
360-
ctx.worker_id, req.success, req.error_message
370+
ctx.worker_id,
371+
req.success,
372+
req.error_message
361373
);
362374

363375
self.send_to_job_queue(
@@ -382,8 +394,17 @@ impl JobControllerGrpc for ControllerServer {
382394
) -> Result<Response<TaskCheckpointEventResp>, Status> {
383395
let req = request.into_inner();
384396

385-
debug!("received task checkpoint event {:?}", req);
386-
let job_id = job_id_from_context(&req.worker_context)?;
397+
let ctx = req
398+
.worker_context
399+
.as_ref()
400+
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
401+
debug!(
402+
job_id = ctx.job_id,
403+
pipeline_id = ctx.pipeline_id,
404+
"received task checkpoint event {:?}",
405+
req
406+
);
407+
let job_id = ctx.job_id.clone();
387408

388409
self.send_to_job_queue(
389410
&job_id,
@@ -400,8 +421,17 @@ impl JobControllerGrpc for ControllerServer {
400421
) -> Result<Response<TaskCheckpointCompletedResp>, Status> {
401422
let req = request.into_inner();
402423

403-
debug!("received task checkpoint completed {:?}", req);
404-
let job_id = job_id_from_context(&req.worker_context)?;
424+
let ctx = req
425+
.worker_context
426+
.as_ref()
427+
.ok_or_else(|| Status::invalid_argument("missing worker_context"))?;
428+
debug!(
429+
job_id = ctx.job_id,
430+
pipeline_id = ctx.pipeline_id,
431+
"received task checkpoint completed {:?}",
432+
req
433+
);
434+
let job_id = ctx.job_id.clone();
405435

406436
self.send_to_job_queue(
407437
&job_id,
@@ -500,6 +530,7 @@ impl JobControllerGrpc for ControllerServer {
500530

501531
info!(
502532
job_id = ctx.job_id,
533+
pipeline_id = ctx.pipeline_id,
503534
operator_id = err.operator_id,
504535
message = "operator error",
505536
error_message = err.error,

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

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ impl Scheduler for EmbeddedScheduler {
5050
let guard = shutdown.guard("embedded-worker");
5151

5252
let job_id = req.job_id.clone();
53+
let pipeline_id = req.pipeline_id.clone();
54+
let log_job_id = job_id.clone();
5355
let generation = req.generation;
5456
let worker_id = WorkerId(self.worker_counter.fetch_add(1, Ordering::SeqCst));
5557
let handle = tokio::task::spawn(async move {
@@ -61,19 +63,38 @@ impl Scheduler for EmbeddedScheduler {
6163
req.generation,
6264
guard,
6365
);
66+
let worker_job_id = log_job_id.clone();
67+
let worker_pipeline_id = pipeline_id.clone();
6468

6569
match tokio::task::spawn(async move {
6670
if let Err(e) = server.start_async().await {
67-
error!("Failed to start worker {:?}: {:?}", worker_id, e);
71+
error!(
72+
job_id = *worker_job_id,
73+
pipeline_id = *worker_pipeline_id,
74+
"Failed to start worker {:?}: {:?}",
75+
worker_id,
76+
e
77+
);
6878
}
6979
})
7080
.await
7181
{
7282
Ok(_) => {
73-
info!("Worker {:?} finished", worker_id);
83+
info!(
84+
job_id = *log_job_id,
85+
pipeline_id = *pipeline_id,
86+
"Worker {:?} finished",
87+
worker_id
88+
);
7489
}
7590
Err(err) => {
76-
error!("Worker {:?} panicked: {:?}", worker_id, err);
91+
error!(
92+
job_id = *log_job_id,
93+
pipeline_id = *pipeline_id,
94+
"Worker {:?} panicked: {:?}",
95+
worker_id,
96+
err
97+
);
7798
}
7899
}
79100
});

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -253,6 +253,7 @@ impl Scheduler for KubernetesScheduler {
253253

254254
info!(
255255
job_id = *req.job_id,
256+
pipeline_id = *req.pipeline_id,
256257
message = "starting workers on k8s",
257258
replicas = pods.len(),
258259
task_slots = req.slots
@@ -261,6 +262,7 @@ impl Scheduler for KubernetesScheduler {
261262
for pod in pods {
262263
info!(
263264
job_id = *req.job_id,
265+
pipeline_id = *req.pipeline_id,
264266
message = "starting worker",
265267
pod = pod.metadata.name
266268
);

0 commit comments

Comments
 (0)