feat(telemetry): propagate request ids through the whole request chain

Modified app/aihost/src/main.rs +15 −2 View app/aihost/src/main.rs

@@ -116,10 +116,23 @@

116116 }

117117 };

118118 let job_id = task.job_id();

119 let span = tracing::info_span!("ai.task", job_id = %job_id, kind = task.kind_str());

119 let span = tracing::info_span!(

120 "ai.task",

121 job_id = %job_id,

122 kind = task.kind_str(),

123 request_id = tracing::field::Empty

124 );

125 if let Some(request_id) = task.request_id() {

126 span.record("request_id", request_id);

127 }

120128 export::telemetry::set_parent_from_traceparent(&span, task.trace());

121129 let started = std::time::Instant::now();

122 match service.run_task(&task).instrument(span).await {

130 match export::middleware::with_request_id(

131 task.request_id(),

132 service.run_task(&task).instrument(span),

133 )

134 .await

135 {

123136 Ok(()) => {

124137 attempts.remove(&job_id);

125138 export::metrics::ai_job(task.kind_str(), "ok");

Modified app/im/src/main.rs +21 −2 View app/im/src/main.rs

@@ -54,13 +54,32 @@

5454 let service = chat.clone();

5555 tokio::spawn(async move {

5656 let io = TokioIo::new(stream);

57 let service = hyper::service::service_fn(move |req| {

57 let service = hyper::service::service_fn(move |mut req| {

5858 let service = service.clone();

59 let request_id = req

60 .headers()

61 .get(export::middleware::REQUEST_ID_HEADER)

62 .and_then(|value| value.to_str().ok())

63 .filter(|value| !value.is_empty() && value.len() <= 128)

64 .map(str::to_owned)

65 .unwrap_or_else(export::middleware::new_request_id);

66 req.extensions_mut()

67 .insert(export::middleware::RequestId(request_id.clone()));

5968 let span = export::telemetry::http_span(&req);

6069 if let Some(traceparent) = query_traceparent(req.uri().query()) {

6170 export::telemetry::set_parent_from_traceparent(&span, Some(traceparent));

6271 }

63 async move { gateway::route(service, req).await }.instrument(span)

72 let response_header = request_id.parse().ok();

73 async move {

74 let mut response = gateway::route(service, req).await;

75 if let (Ok(response), Some(header)) = (&mut response, response_header) {

76 response

77 .headers_mut()

78 .insert(export::middleware::REQUEST_ID_HEADER, header);

79 }

80 response

81 }

82 .instrument(span)

6483 });

6584 if let Err(error) = hyper::server::conn::http1::Builder::new()

6685 .serve_connection(io, service)

Modified app/runner/src/actions.rs +8 −1 View app/runner/src/actions.rs

@@ -102,11 +102,18 @@

102102 let renew_client = client.clone();

103103 let job_id = payload.job_id;

104104 let interval = payload.lease.heartbeat_interval_secs.max(1);

105 let request_id = export::telemetry::request_id();

105106 let renew = tokio::spawn(async move {

106107 let mut ticker = tokio::time::interval(std::time::Duration::from_secs(interval));

107108 loop {

108109 ticker.tick().await;

109 if renew_client.renew(&job_id).await.is_err() {

110 if export::middleware::with_request_id(

111 request_id.as_deref(),

112 renew_client.renew(&job_id),

113 )

114 .await

115 .is_err()

116 {

110117 crate::executor::request_cancel(&[job_id]);

111118 }

112119 }

Modified app/runner/src/main.rs +69 −41 View app/runner/src/main.rs

@@ -171,9 +171,17 @@

171171 job_id = %payload.job_id,

172172 run_id = %payload.run_id,

173173 job = %payload.job_name,

174 request_id = tracing::field::Empty

174175 );

176 if let Some(request_id) = payload.request_id.as_deref() {

177 span.record("request_id", request_id);

178 }

175179 export::telemetry::set_parent_from_traceparent(&span, payload.trace.as_deref());

176 run_claimed_job(client, cfg, payload).instrument(span).await;

180 export::middleware::with_request_id(

181 payload.request_id.as_deref(),

182 run_claimed_job(client, cfg, payload).instrument(span),

183 )

184 .await;

177185}

178186

179187async fn run_claimed_job(

@@ -228,6 +236,7 @@

228236 warn!(%error, "job start report failed");

229237 }

230238

239 let request_id = export::telemetry::request_id();

231240 let seq = std::sync::Arc::new(std::sync::atomic::AtomicI32::new(0));

232241 let log_client = client.clone();

233242 let log_job = payload.job_id;

@@ -237,30 +246,36 @@

237246 let on_log = {

238247 let seq = std::sync::Arc::clone(&seq);

239248 let secrets = log_secrets.clone();

249 let log_request_id = request_id.clone();

240250 move |text: String| {

241251 let client = log_client.clone();

242252 let job_id = log_job;

243253 let text = mask_secrets(text, &secrets);

244254 let num = seq.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

255 let request_id = log_request_id.clone();

245256 tokio::spawn(async move {

246 let _ = client

247 .upload_logs(

257 let _ = export::middleware::with_request_id(

258 request_id.as_deref(),

259 client.upload_logs(

248260 job_id,

249261 &LogChunksRequest {

250262 chunks: vec![LogChunk { seq: num, text }],

251263 },

252 )

253 .await;

264 ),

265 )

266 .await;

254267 });

255268 }

256269 };

257270

258271 let step_client = client.clone();

259272 let step_job = payload.job_id;

273 let step_request_id = request_id.clone();

260274 let on_status =

261275 move |index: i32, signal: runner::executor::StepSignal, exit_code: Option<i32>| {

262276 let client = step_client.clone();

263277 let job_id = step_job;

278 let request_id = step_request_id.clone();

264279 tokio::spawn(async move {

265280 let report = StepReport {

266281 state: match signal {

@@ -279,62 +294,75 @@

279294 .then(time::OffsetDateTime::now_utc),

280295 ..Default::default()

281296 };

282 let _ = client.report_step(job_id, index, &report).await;

297 let _ = export::middleware::with_request_id(

298 request_id.as_deref(),

299 client.report_step(job_id, index, &report),

300 )

301 .await;

283302 });

284303 };

285304

286305 let artifact_client = client.clone();

287306 let artifact_job = payload.job_id;

307 let artifact_request_id = request_id.clone();

288308 let on_artifact = move |name: String, path: String, size: i64, bytes: Vec<u8>| {

289309 let client = artifact_client.clone();

290310 let job_id = artifact_job;

311 let request_id = artifact_request_id.clone();

291312 tokio::spawn(async move {

292 let sha256 = runner::executor::sha256_hex(&bytes);

293 let init = match client

294 .init_artifact(

295 job_id,

296 &ArtifactInitRequest {

297 name: name.clone(),

298 size,

299 sha256: Some(sha256.clone()),

300 },

301 )

302 .await

303 {

304 Ok(init) => init,

305 Err(error) => {

306 warn!(%error, %name, "artifact init failed");

313 export::middleware::with_request_id(request_id.as_deref(), async move {

314 let sha256 = runner::executor::sha256_hex(&bytes);

315 let init = match client

316 .init_artifact(

317 job_id,

318 &ArtifactInitRequest {

319 name: name.clone(),

320 size,

321 sha256: Some(sha256.clone()),

322 },

323 )

324 .await

325 {

326 Ok(init) => init,

327 Err(error) => {

328 warn!(%error, %name, "artifact init failed");

329 return;

330 }

331 };

332 if let Err(error) = runner::executor::upload_file(&init.upload_url, bytes).await {

333 warn!(%error, %name, "artifact upload failed");

307334 return;

308335 }

309 };

310 if let Err(error) = runner::executor::upload_file(&init.upload_url, bytes).await {

311 warn!(%error, %name, "artifact upload failed");

312 return;

313 }

314 let _ = path;

315 if let Err(error) = client

316 .complete_artifact(

317 job_id,

318 init.artifact_id,

319 &ArtifactCompleteRequest {

320 size,

321 sha256: Some(sha256),

322 },

323 )

324 .await

325 {

326 warn!(%error, %name, "artifact completion failed");

327 }

336 let _ = path;

337 if let Err(error) = client

338 .complete_artifact(

339 job_id,

340 init.artifact_id,

341 &ArtifactCompleteRequest {

342 size,

343 sha256: Some(sha256),

344 },

345 )

346 .await

347 {

348 warn!(%error, %name, "artifact completion failed");

349 }

350 })

351 .await;

328352 });

329353 };

330354

331355 let renew_client = client.clone();

332356 let renew_job = payload.job_id;

357 let renew_request_id = request_id;

333358 let renew = move || {

334359 let client = renew_client.clone();

335360 let job_id = renew_job;

361 let request_id = renew_request_id.clone();

336362 tokio::spawn(async move {

337 let _ = client.renew(&job_id).await;

363 let _ =

364 export::middleware::with_request_id(request_id.as_deref(), client.renew(&job_id))

365 .await;

338366 });

339367 };

340368

Modified app/webhook/src/main.rs +16 −3 View app/webhook/src/main.rs

@@ -82,7 +82,8 @@

8282 service.state.redis.clone(),

8383 ))

8484 .layer(export::telemetry::http_trace_layer())

85 .layer(export::telemetry::metrics_layer());

85 .layer(export::telemetry::metrics_layer())

86 .layer(export::middleware::AppTraceLayer::new());

8687 info!(%addr, "webhook scheduler listening");

8788

8889 let server_cancel = cancel.clone();

@@ -151,9 +152,21 @@

151152 ci_protocol::events::CiEvent::PullRequest(pr) => pr.id,

152153 ci_protocol::events::CiEvent::Trigger(event) => event.id,

153154 };

154 let span = tracing::info_span!("ci.event", event = %event_id);

155 let span = tracing::info_span!(

156 "ci.event",

157 event = %event_id,

158 request_id = tracing::field::Empty

159 );

160 if let Some(request_id) = event.request_id() {

161 span.record("request_id", request_id);

162 }

155163 export::telemetry::set_parent_from_traceparent(&span, event.trace());

156 match service.handle_event(&event).instrument(span).await {

164 match export::middleware::with_request_id(

165 event.request_id(),

166 service.handle_event(&event).instrument(span),

167 )

168 .await

169 {

157170 Ok(created) => {

158171 attempts.remove(&event_id);

159172 export::metrics::ci_runs_created(event_kind, created.len());

Modified lib/chat/src/services/messages.rs +6 −1 View lib/chat/src/services/messages.rs

@@ -105,6 +105,7 @@

105105 actor_id: Some(actor),

106106 notification_id: None,

107107 trace: export::telemetry::traceparent(),

108 request_id: export::telemetry::request_id(),

108109 },

109110 )

110111 .await;

@@ -369,7 +370,11 @@

369370 Ok(())

370371 }

371372

372 #[tracing::instrument(name = "chat.send_message", skip_all)]

373 #[tracing::instrument(

374 name = "chat.send_message",

375 skip_all,

376 fields(caller = %caller, room_id = %room_id)

377)]

373378 pub async fn send_message(

374379 &self,

375380 caller: Uuid,

Modified lib/ci/src/actions_events.rs +2 −0 View lib/ci/src/actions_events.rs

@@ -46,6 +46,7 @@

4646 event: "repository_dispatch".into(),

4747 payload: json!({"action":command.event_type,"client_payload":command.client_payload}),

4848 trace: export::telemetry::traceparent(),

49 request_id: export::telemetry::request_id(),

4950 };

5051 self.handle_trigger_event(&event).await

5152 }

@@ -352,6 +353,7 @@

352353 event: "workflow_run".into(),

353354 payload: json!({"action":action,"_workflow_run_depth":run.event_payload["_workflow_run_depth"].as_u64().unwrap_or(0)+1,"workflow_run":{"id":run.actions_id,"name":run.workflow_name,"head_branch":run.branch,"head_sha":run.commit_sha,"status":if action=="completed"{"completed"}else if action=="requested"{"queued"}else{"in_progress"},"conclusion":if action=="completed"{Some(crate::share::states::run_state_str(run.state))}else{None},"event":run.event.as_str()}}),

354355 trace: export::telemetry::traceparent(),

356 request_id: export::telemetry::request_id(),

355357 });

356358 crate::events::publish_event(

357359 &self.state.redis,

Modified lib/ci/src/actions_payload.rs +5 −0 View lib/ci/src/actions_payload.rs

@@ -129,6 +129,11 @@

129129 .get("_gitdata_trace")

130130 .and_then(|value| value.as_str())

131131 .map(str::to_owned),

132 request_id: run

133 .event_payload

134 .get("_gitdata_request_id")

135 .and_then(|value| value.as_str())

136 .map(str::to_owned),

132137 actions: Some(actions),

133138 })

134139 }

Modified lib/ci/src/runners.rs +5 −0 View lib/ci/src/runners.rs

@@ -504,6 +504,11 @@

504504 .get("_gitdata_trace")

505505 .and_then(|value| value.as_str())

506506 .map(str::to_owned),

507 request_id: run

508 .event_payload

509 .get("_gitdata_request_id")

510 .and_then(|value| value.as_str())

511 .map(str::to_owned),

507512 actions: None,

508513 };

509514 Ok(payload)

Modified lib/ci/src/runs.rs +2 −0 View lib/ci/src/runs.rs

@@ -69,6 +69,7 @@

6969 "actor_id": event.actor_id,

7070 "event_id": event.id,

7171 "_gitdata_trace": export::telemetry::traceparent(),

72 "_gitdata_request_id": export::telemetry::request_id(),

7273 }),

7374 commit_sha: update.new_oid.clone(),

7475 git_ref: update.name.clone(),

@@ -97,6 +98,7 @@

9798 "event_id": event.id,

9899 "pull_request":{"number":event.number,"merged":event.action==ci_protocol::events::PullRequestAction::Merged,"head":{"sha":event.head_sha,"ref":event.head_ref,"repo":{"fork":event.head_repo_id.is_some_and(|id|id!=event.repo_id)}},"base":{"sha":event.base_sha,"ref":event.base_ref}},

99100 "_gitdata_trace": export::telemetry::traceparent(),

101 "_gitdata_request_id": export::telemetry::request_id(),

100102 }),

101103 commit_sha: event.head_sha.clone(),

102104 git_ref: format!("refs/pull/{}/head", event.number),

Modified lib/ci-protocol/src/ai.rs +18 −0 View lib/ci-protocol/src/ai.rs

@@ -18,6 +18,8 @@

1818 language: Option<String>,

1919 #[serde(default)]

2020 trace: Option<String>,

21 #[serde(default)]

22 request_id: Option<String>,

2123 },

2224 CommitReview {

2325 job_id: Uuid,

@@ -27,6 +29,8 @@

2729 language: Option<String>,

2830 #[serde(default)]

2931 trace: Option<String>,

32 #[serde(default)]

33 request_id: Option<String>,

3034 },

3135 RepoSummary {

3236 job_id: Uuid,

@@ -38,6 +42,8 @@

3842 language: Option<String>,

3943 #[serde(default)]

4044 trace: Option<String>,

45 #[serde(default)]

46 request_id: Option<String>,

4147 },

4248 ReleaseNotes {

4349 job_id: Uuid,

@@ -50,6 +56,8 @@

5056 language: Option<String>,

5157 #[serde(default)]

5258 trace: Option<String>,

59 #[serde(default)]

60 request_id: Option<String>,

5361 },

5462}

5563

@@ -134,4 +142,14 @@

134142 | Self::ReleaseNotes { trace, .. } => trace.as_deref(),

135143 }

136144 }

145

146 #[must_use]

147 pub fn request_id(&self) -> Option<&str> {

148 match self {

149 Self::PullReview { request_id, .. }

150 | Self::CommitReview { request_id, .. }

151 | Self::RepoSummary { request_id, .. }

152 | Self::ReleaseNotes { request_id, .. } => request_id.as_deref(),

153 }

154 }

137155}

Modified lib/ci-protocol/src/events.rs +15 −0 View lib/ci-protocol/src/events.rs

@@ -75,6 +75,8 @@

7575 /// Producing trace context, carried across the queue to webhook consumers.

7676 #[serde(default)]

7777 pub trace: Option<String>,

78 #[serde(default)]

79 pub request_id: Option<String>,

7880}

7981

8082#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

@@ -93,6 +95,8 @@

9395 pub head_repo_id: Option<Uuid>,

9496 #[serde(default)]

9597 pub trace: Option<String>,

98 #[serde(default)]

99 pub request_id: Option<String>,

96100}

97101

98102#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

@@ -105,6 +109,8 @@

105109 pub payload: serde_json::Value,

106110 #[serde(default)]

107111 pub trace: Option<String>,

112 #[serde(default)]

113 pub request_id: Option<String>,

108114}

109115

110116impl CiEvent {

@@ -116,4 +122,13 @@

116122 Self::Trigger(event) => event.trace.as_deref(),

117123 }

118124 }

125

126 #[must_use]

127 pub fn request_id(&self) -> Option<&str> {

128 match self {

129 Self::Push(event) => event.request_id.as_deref(),

130 Self::PullRequest(event) => event.request_id.as_deref(),

131 Self::Trigger(event) => event.request_id.as_deref(),

132 }

133 }

119134}

Modified lib/ci-protocol/src/runner.rs +3 −0 View lib/ci-protocol/src/runner.rs

@@ -139,6 +139,9 @@

139139 /// continue the run's trace in its own process.

140140 #[serde(default, skip_serializing_if = "Option::is_none")]

141141 pub trace: Option<String>,

142 /// Request id of the CI event that produced the run.

143 #[serde(default, skip_serializing_if = "Option::is_none")]

144 pub request_id: Option<String>,

142145 #[serde(default, skip_serializing_if = "Option::is_none")]

143146 pub actions: Option<crate::actions::ActionsExecution>,

144147}

Modified lib/export/src/middleware.rs +23 −1 View lib/export/src/middleware.rs

@@ -15,12 +15,34 @@

1515

1616const MAX_INBOUND_ID_LEN: usize = 128;

1717

18tokio::task_local! {

19 static REQUEST_ID: String;

20}

21

1822#[derive(Debug, Clone)]

1923pub struct RequestId(pub String);

2024

2125#[derive(Debug, Clone)]

2226pub struct LogId(pub String);

2327

28/// The request id of the request currently being handled, when one is set.

29pub fn request_id() -> Option<String> {

30 REQUEST_ID.try_with(|id| id.clone()).ok()

31}

32

33/// Runs `future` with `request_id` available to nested code, so queue

34/// consumers keep the id of the request that produced the message.

35pub async fn with_request_id<F: Future>(request_id: Option<&str>, future: F) -> F::Output {

36 match request_id {

37 Some(request_id) => REQUEST_ID.scope(request_id.to_string(), future).await,

38 None => future.await,

39 }

40}

41

42pub fn new_request_id() -> String {

43 new_id()

44}

45

2446#[derive(Debug, Clone, Copy, Default)]

2547pub struct AppTraceLayer;

2648

@@ -63,7 +85,7 @@

6385 req.extensions_mut().insert(RequestId(request_id.clone()));

6486 req.extensions_mut().insert(LogId(log_id.clone()));

6587

66 let fut = self.inner.call(req);

88 let fut = REQUEST_ID.scope(request_id.clone(), self.inner.call(req));

6789 Box::pin(async move {

6890 let mut res = fut.await?;

6991 if let Ok(value) = HeaderValue::try_from(request_id.as_str()) {

Modified lib/export/src/push.rs +3 −0 View lib/export/src/push.rs

@@ -100,6 +100,8 @@

100100 pub notification_id: Option<Uuid>,

101101 #[serde(default)]

102102 pub trace: Option<String>,

103 #[serde(default)]

104 pub request_id: Option<String>,

103105}

104106

105107impl PushJob {

@@ -133,6 +135,7 @@

133135 actor_id,

134136 notification_id,

135137 trace: crate::telemetry::traceparent(),

138 request_id: crate::telemetry::request_id(),

136139 })

137140 }

138141

Modified lib/export/src/telemetry.rs +10 −0 View lib/export/src/telemetry.rs

@@ -168,6 +168,11 @@

168168 .map(str::to_owned)

169169}

170170

171/// Serialises the request id for a queue payload.

172pub fn request_id() -> Option<String> {

173 crate::middleware::request_id()

174}

175

171176pub fn http_span<B>(request: &Request<B>) -> Span {

172177 let method = request.method().as_str();

173178 let route = request

@@ -309,6 +314,11 @@

309314 global::get_text_map_propagator(|propagator| {

310315 propagator.inject_context(&span.context(), &mut HeaderInjector(&mut headers));

311316 });

317 if let Some(request_id) = crate::middleware::request_id()

318 && let Ok(value) = request_id.parse()

319 {

320 headers.insert(crate::middleware::REQUEST_ID_HEADER, value);

321 }

312322

313323 let result = request

314324 .headers(headers)

Modified lib/git/src/service/ai_review.rs +1 −0 View lib/git/src/service/ai_review.rs

@@ -114,6 +114,7 @@

114114 }

115115 },

116116 trace: export::telemetry::traceparent(),

117 request_id: export::telemetry::request_id(),

117118 };

118119 let job_id =

119120 crate::sync::ai_task::submit_ai_task(self.db_write(), &self.state.redis, cfg, &task)

Modified lib/git/src/service/ai_summary.rs +1 −0 View lib/git/src/service/ai_summary.rs

@@ -168,6 +168,7 @@

168168 None => crate::sync::ai_language::actor_language(self.db_read(), actor).await,

169169 },

170170 trace: export::telemetry::traceparent(),

171 request_id: export::telemetry::request_id(),

171172 };

172173 crate::sync::ai_task::submit_ai_task(self.db_write(), &self.state.redis, cfg, &task)

173174 .await

Modified lib/git/src/service/releases.rs +1 −0 View lib/git/src/service/releases.rs

@@ -287,6 +287,7 @@

287287 to_sha: release.sha.clone(),

288288 language: crate::sync::ai_language::actor_language(self.db_read(), actor).await,

289289 trace: export::telemetry::traceparent(),

290 request_id: export::telemetry::request_id(),

290291 };

291292 crate::sync::ai_task::submit_ai_task(self.db_write(), &self.state.redis, cfg, &task)

292293 .await

Modified lib/git/src/smart/handler.rs +1 −0 View lib/git/src/smart/handler.rs

@@ -26,6 +26,7 @@

2626 .with_state(service)

2727 .layer(export::telemetry::http_trace_layer())

2828 .layer(export::telemetry::metrics_layer())

29 .layer(export::middleware::AppTraceLayer::new())

2930}

3031

3132async fn dispatch(

Modified lib/git/src/sync/ai_task.rs +7 −0 View lib/git/src/sync/ai_task.rs

@@ -134,6 +134,7 @@

134134 commit_sha: new.clone(),

135135 language: language.clone(),

136136 trace: export::telemetry::traceparent(),

137 request_id: export::telemetry::request_id(),

137138 };

138139 submit_ai_task_logged(conn, redis, cfg, &task).await;

139140 }

@@ -151,6 +152,7 @@

151152 limit: None,

152153 language: language.clone(),

153154 trace: export::telemetry::traceparent(),

155 request_id: export::telemetry::request_id(),

154156 };

155157 submit_ai_task_logged(conn, redis, cfg, &task).await;

156158 }

@@ -246,6 +248,7 @@

246248 to_sha: target_sha.to_string(),

247249 language: language.or(resolve_language_for(db, actor_id).await),

248250 trace: export::telemetry::traceparent(),

251 request_id: export::telemetry::request_id(),

249252 };

250253 let job_id = submit_ai_task(db, redis, cfg, &task).await?;

251254 Ok(Some(job_id))

@@ -323,6 +326,7 @@

323326 head_sha: target_ref,

324327 language,

325328 trace: None,

329 request_id: None,

326330 }),

327331 AiJobKind::CommitReview => Some(ci_protocol::ai::AgentTask::CommitReview {

328332 job_id: row.id,

@@ -330,6 +334,7 @@

330334 commit_sha: target_ref,

331335 language,

332336 trace: None,

337 request_id: None,

333338 }),

334339 AiJobKind::RepoSummary => Some(ci_protocol::ai::AgentTask::RepoSummary {

335340 job_id: row.id,

@@ -339,6 +344,7 @@

339344 limit: None,

340345 language,

341346 trace: None,

347 request_id: None,

342348 }),

343349 AiJobKind::ReleaseNotes => {

344350 let release = repo_release::Entity::find_by_id(row.target_id?)

@@ -354,6 +360,7 @@

354360 to_sha: release.sha,

355361 language,

356362 trace: None,

363 request_id: None,

357364 })

358365 }

359366 }

Modified lib/git/src/sync/event.rs +3 −0 View lib/git/src/sync/event.rs

@@ -49,6 +49,8 @@

4949 pub refs: Vec<PostReceiveRefUpdate>,

5050 #[serde(default)]

5151 pub trace: Option<String>,

52 #[serde(default)]

53 pub request_id: Option<String>,

5254}

5355

5456impl PostReceiveTask {

@@ -66,6 +68,7 @@

6668 actor_id,

6769 refs,

6870 trace: export::telemetry::traceparent(),

71 request_id: export::telemetry::request_id(),

6972 }

7073 }

7174}

Modified lib/git/src/sync/post_receive.rs +1 −0 View lib/git/src/sync/post_receive.rs

@@ -224,6 +224,7 @@

224224 actor_id,

225225 refs: ci_refs,

226226 trace: export::telemetry::traceparent(),

227 request_id: export::telemetry::request_id(),

227228 });

228229 match serde_json::to_vec(&event) {

229230 Ok(payload) => {

Modified lib/git/src/sync/worker.rs +16 −2 View lib/git/src/sync/worker.rs

@@ -172,7 +172,11 @@

172172 task_id = %task.id,

173173 repo_id = %task.repo_id,

174174 refs = task.refs.len(),

175 request_id = tracing::field::Empty

175176 );

177 if let Some(request_id) = task.request_id.as_deref() {

178 span.record("request_id", request_id);

179 }

176180 export::telemetry::set_parent_from_traceparent(&span, task.trace.as_deref());

177181

178182 let repo = match repos::Entity::find_by_id(task.repo_id).one(db.read()).await {

@@ -207,7 +211,12 @@

207211 settings.clone(),

208212 );

209213

210 match rsync.sync().instrument(span.clone()).await {

214 match export::middleware::with_request_id(

215 task.request_id.as_deref(),

216 rsync.sync().instrument(span.clone()),

217 )

218 .await

219 {

211220 Ok(()) => {

212221 export::metrics::sync_job("ok");

213222 info!(

@@ -229,7 +238,12 @@

229238 "pull request refresh failed"

230239 );

231240 }

232 if let Err(e) = webhook.dispatch(&task).instrument(span.clone()).await {

241 if let Err(e) = export::middleware::with_request_id(

242 task.request_id.as_deref(),

243 webhook.dispatch(&task).instrument(span.clone()),

244 )

245 .await

246 {

233247 warn!(

234248 consumer = id,

235249 task_id = %task.id,

Modified lib/pulls/src/services/events.rs +2 −0 View lib/pulls/src/services/events.rs

@@ -143,6 +143,7 @@

143143 base_ref: pull.base_branch.clone(),

144144 head_repo_id: pull.head_repo_id,

145145 trace: export::telemetry::traceparent(),

146 request_id: export::telemetry::request_id(),

146147 });

147148 let payload = match serde_json::to_vec(&event) {

148149 Ok(payload) => payload,

@@ -201,6 +202,7 @@

201202 // 评审语言跟随触发者（推送/开 PR 的人）。

202203 language: git::sync::actor_language(self.db_read(), actor).await,

203204 trace: export::telemetry::traceparent(),

205 request_id: export::telemetry::request_id(),

204206 };

205207 git::sync::ai_task::submit_ai_task_logged(self.db_write(), &self.state.redis, cfg, &task)

206208 .await;

Modified lib/push/src/worker.rs +15 −2 View lib/push/src/worker.rs

@@ -83,9 +83,22 @@

8383 }

8484 };

8585

86 let span = tracing::info_span!("push.delivery", user = %job.user_id, category = job.category.as_str());

86 let span = tracing::info_span!(

87 "push.delivery",

88 user = %job.user_id,

89 category = job.category.as_str(),

90 request_id = tracing::field::Empty

91 );

92 if let Some(request_id) = job.request_id.as_deref() {

93 span.record("request_id", request_id);

94 }

8795 export::telemetry::set_parent_from_traceparent(&span, job.trace.as_deref());

88 if let Err(error) = self.handle(&job).instrument(span).await {

96 if let Err(error) = export::middleware::with_request_id(

97 job.request_id.as_deref(),

98 self.handle(&job).instrument(span),

99 )

100 .await

101 {

89102 tracing::warn!(

90103 %error,

91104 user = %job.user_id,

Modified lib/sandbox-rpc/src/lib.rs +6 −0 View lib/sandbox-rpc/src/lib.rs

@@ -17,6 +17,7 @@

1717

1818pub const TRUSTED_KEY_HEADER: &str = "x-trusted-key";

1919pub const TRACEPARENT_HEADER: &str = "traceparent";

20pub const REQUEST_ID_HEADER: &str = "x-request-id";

2021pub const SCOPE_STREAM: &str = "stream";

2122

2223#[derive(Debug, thiserror::Error)]

@@ -77,6 +78,11 @@

7778 {

7879 request.metadata_mut().insert(TRACEPARENT_HEADER, value);

7980 }

81 if let Some(request_id) = export::middleware::request_id()

82 && let Ok(value) = MetadataValue::try_from(request_id.as_str())

83 {

84 request.metadata_mut().insert(REQUEST_ID_HEADER, value);

85 }

8086 Ok(request)

8187 }

8288}

Modified lib/sandbox-rpc/Cargo.toml +1 −0 View lib/sandbox-rpc/Cargo.toml

@@ -5,6 +5,7 @@

55

66[dependencies]

77config = { workspace = true }

8export = { workspace = true }

89opentelemetry = { workspace = true }

910prost = { workspace = true }

1011thiserror = { workspace = true }

Modified Cargo.lock +1 −0 View Cargo.lock

@@ -8318,6 +8318,7 @@

83188318version = "1.0.0"

83198319dependencies = [

83208320 "config",

8321 "export",

83218322 "opentelemetry",

83228323 "prost 0.14.4",

83238324 "thiserror 2.0.20",