feat(telemetry): carry trace context through queues and the CI runner

Modified app/aihost/src/main.rs +4 −1 View app/aihost/src/main.rs

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

77use git::sync::queue::SyncQueue;

88use tokio::net::TcpListener;

99use tokio_util::sync::CancellationToken;

10use tracing::Instrument as _;

1011use tracing::{info, warn};

1112

1213#[tokio::main]

@@ -105,7 +106,9 @@

105106 }

106107 };

107108 let job_id = task.job_id();

108 match service.run_task(&task).await {

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

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

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

109112 Ok(()) => {

110113 attempts.remove(&job_id);

111114 info!(%job_id, kind = task.kind_str(), "ai task completed");

Modified app/im/src/main.rs +9 −0 View app/im/src/main.rs

@@ -53,6 +53,9 @@

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

5454 let service = service.clone();

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

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

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

58 }

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

5760 });

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

@@ -70,3 +73,9 @@

7073 info!("im shut down");

7174 Ok(())

7275}

76

77fn query_traceparent(query: Option<&str>) -> Option<&str> {

78 query?

79 .split('&')

80 .find_map(|pair| pair.strip_prefix("traceparent="))

81}

Modified app/mails/src/main.rs +2 −7 View app/mails/src/main.rs

@@ -9,7 +9,7 @@

99

1010#[tokio::main]

1111async fn main() -> anyhow::Result<()> {

12 init_tracing();

12 export::telemetry::init_tracing();

1313

1414 let cfg = MailsConfig::load().context("loading configuration")?;

1515

@@ -69,12 +69,7 @@

6969

7070 result.map_err(|e| anyhow::anyhow!(e.to_string()))?;

7171

72 export::telemetry::shutdown_telemetry();

7273 info!("mails worker exited cleanly");

7374 Ok(())

74}

75

76fn init_tracing() {

77 use tracing_subscriber::EnvFilter;

78 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));

79 tracing_subscriber::fmt().with_env_filter(filter).init();

8075}

Modified app/runner/src/client.rs +6 −2 View app/runner/src/client.rs

@@ -44,11 +44,15 @@

4444 path: &str,

4545 body: &B,

4646 ) -> anyhow::Result<T> {

47 let mut request = self.http.post(self.url(path)).json(body);

47 let url = self.url(path);

48 let mut request = self.http.post(&url).json(body);

4849 if let Some(credential) = &self.credential {

4950 request = request.bearer_auth(credential);

5051 }

51 let response = request.send().await.context("sending runner request")?;

52 let span = export::telemetry::http_client_span("POST", &url);

53 let response = export::telemetry::send(span, request)

54 .await

55 .context("sending runner request")?;

5256 let status = response.status();

5357 let text = response.text().await.unwrap_or_default();

5458 if status != StatusCode::OK {

Modified app/runner/src/executor.rs +8 −2 View app/runner/src/executor.rs

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

77use config::runner::RunnerConfig;

88use tokio::io::{AsyncBufReadExt, BufReader};

99use tokio::process::Command;

10use tracing::Instrument as _;

1011

1112pub const MAX_STEP_OUTPUT_BYTES: usize = 4 * 1024 * 1024;

1213

@@ -147,7 +148,10 @@

147148 step.index + 1,

148149 step.name

149150 ));

150 let result = run_step(cfg, payload, step, &repo_dir, hooks.clone(), deadline).await;

151 let step_span = tracing::info_span!("ci.step", index = step.index, name = %step.name);

152 let result = run_step(cfg, payload, step, &repo_dir, hooks.clone(), deadline)

153 .instrument(step_span)

154 .await;

151155 match result {

152156 Ok(code) => {

153157 (hooks.on_status)(step.index, StepSignal::Finished, Some(code));

@@ -572,7 +576,9 @@

572576}

573577

574578pub async fn upload_file(url: &str, bytes: Vec<u8>) -> anyhow::Result<()> {

575 let response = reqwest::Client::new().put(url).body(bytes).send().await?;

579 let span = export::telemetry::http_client_span("PUT", url);

580 let response =

581 export::telemetry::send(span, reqwest::Client::new().put(url).body(bytes)).await?;

576582 if !response.status().is_success() {

577583 anyhow::bail!("upload failed with {}", response.status());

578584 }

Modified app/runner/src/main.rs +18 −4 View app/runner/src/main.rs

@@ -7,7 +7,7 @@

77use runner::client::RunnerClient;

88use runner::executor::{JobOutcome, StepHooks, execute_job};

99use runner::state::{credential_path, load_credential, store_credential};

10use tracing::{info, warn};

10use tracing::{Instrument as _, info, warn};

1111

1212#[tokio::main]

1313async fn main() -> anyhow::Result<()> {

@@ -157,6 +157,7 @@

157157 }

158158

159159 heartbeat_task.abort();

160 export::telemetry::shutdown_telemetry();

160161 Ok(())

161162}

162163

@@ -165,6 +166,21 @@

165166 cfg: &config::runner::RunnerConfig,

166167 payload: &JobPayload,

167168) {

169 let span = tracing::info_span!(

170 "ci.job",

171 job_id = %payload.job_id,

172 run_id = %payload.run_id,

173 job = %payload.job_name,

174 );

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

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

177}

178

179async fn run_claimed_job(

180 client: &RunnerClient,

181 cfg: &config::runner::RunnerConfig,

182 payload: &JobPayload,

183) {

168184 let finish = if payload.actions.is_some() {

169185 runner::actions::execute(client, cfg, payload).await

170186 } else {

@@ -346,7 +362,5 @@

346362}

347363

348364fn init_tracing() {

349 use tracing_subscriber::EnvFilter;

350 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));

351 tracing_subscriber::fmt().with_env_filter(filter).init();

365 export::telemetry::init_tracing();

352366}

Modified app/runner/Cargo.toml +1 −0 View app/runner/Cargo.toml

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

77[dependencies]

88ci-protocol = { workspace = true }

99config = { workspace = true }

10export = { workspace = true }

1011serde = { workspace = true, features = ["derive"] }

1112serde_json = { workspace = true }

1213tokio = { workspace = true, features = ["signal", "macros", "rt-multi-thread", "time", "sync", "process", "fs"] }

Modified app/webhook/src/main.rs +4 −1 View app/webhook/src/main.rs

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

77use config::scoped::{ScopedConfig, WebhookServiceConfig};

88use tokio::net::TcpListener;

99use tokio_util::sync::CancellationToken;

10use tracing::Instrument as _;

1011use tracing::{info, warn};

1112

1213#[tokio::main]

@@ -132,7 +133,9 @@

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

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

134135 };

135 match service.handle_event(&event).await {

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

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

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

136139 Ok(created) => {

137140 attempts.remove(&event_id);

138141 if !created.is_empty() {

Modified charts/act/templates/deployment.yaml +6 −0 View charts/act/templates/deployment.yaml

@@ -120,6 +120,12 @@

120120 value: {{ .Values.runner.actionsEngineBinary | quote }}

121121 - name: APP_RUNNER__ACTIONS_CACHE_DIR

122122 value: {{ .Values.runner.actionsCacheDir | quote }}

123 {{- if .Values.metrics.otlpEndpoint }}

124 - name: OTEL_EXPORTER_OTLP_ENDPOINT

125 value: {{ .Values.metrics.otlpEndpoint | quote }}

126 - name: OTEL_SERVICE_NAME

127 value: "runner-pool"

128 {{- end }}

123129 {{- with .Values.runner.extraEnv }}

124130 {{- toYaml . | nindent 12 }}

125131 {{- end }}

Modified charts/act/values.yaml +4 −0 View charts/act/values.yaml

@@ -8,6 +8,10 @@

88# <service>.<namespace>.svc.cluster.local:<port>

99serverUrl: "http://gitdata-webhook.gitdata.svc.cluster.local:3005"

1010

11# OTLP 导出端点（与主 chart 的 metrics.otlpEndpoint 对齐；留空则只输出日志）。

12metrics:

13 otlpEndpoint: ""

14

1115image:

1216 repository: harbor.gitdata.me/app/gitdata

1317 tag: ""

Modified charts/core/templates/runner.yaml +6 −0 View charts/core/templates/runner.yaml

@@ -73,6 +73,12 @@

7373 value: /opt/gitdata/actions/bin/gitdata-actions-engine

7474 - name: APP_RUNNER__ACTIONS_CACHE_DIR

7575 value: /var/lib/gitdata-runner/actions

76 {{- if .Values.metrics.otlpEndpoint }}

77 - name: OTEL_EXPORTER_OTLP_ENDPOINT

78 value: {{ .Values.metrics.otlpEndpoint | quote }}

79 - name: OTEL_SERVICE_NAME

80 value: "runner"

81 {{- end }}

7682 {{- with .Values.runner.extraEnv }}

7783 {{- toYaml . | nindent 12 }}

7884 {{- end }}

Modified charts/prod/values/runner-prod.yaml +3 −0 View charts/prod/values/runner-prod.yaml

@@ -39,6 +39,9 @@

3939 requests: { cpu: 500m, memory: 512Mi }

4040 limits: { memory: 4Gi }

4141

42metrics:

43 otlpEndpoint: "http://otel-collector.monitoring.svc.cluster.local:4318"

44

4245autoscaling:

4346 enabled: true

4447 minReplicas: 3

Modified lib/agent/src/service.rs +4 −0 View lib/agent/src/service.rs

@@ -148,6 +148,7 @@

148148 pull_request_id,

149149 head_sha,

150150 language,

151 ..

151152 } = task

152153 else {

153154 return Err(AgentError::InvalidTask("not a pull review".to_string()));

@@ -316,6 +317,7 @@

316317 repo_id,

317318 commit_sha,

318319 language,

320 ..

319321 } = task

320322 else {

321323 return Err(AgentError::InvalidTask("not a commit review".to_string()));

@@ -457,6 +459,7 @@

457459 base_sha,

458460 limit,

459461 language,

462 ..

460463 } = task

461464 else {

462465 return Err(AgentError::InvalidTask("not a repo summary".to_string()));

@@ -599,6 +602,7 @@

599602 from_sha,

600603 to_sha,

601604 language,

605 ..

602606 } = task

603607 else {

604608 return Err(AgentError::InvalidTask("not a release notes".to_string()));

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

@@ -104,6 +104,7 @@

104104 })),

105105 actor_id: Some(actor),

106106 notification_id: None,

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

107108 },

108109 )

109110 .await;

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

@@ -45,6 +45,7 @@

4545 actor_id: Some(caller),

4646 event: "repository_dispatch".into(),

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

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

4849 };

4950 self.handle_trigger_event(&event).await

5051 }

@@ -348,6 +349,7 @@

348349 actor_id: run.trigger_user_id,

349350 event: "workflow_run".into(),

350351 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()}}),

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

351353 });

352354 crate::events::publish_event(

353355 &self.state.redis,

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

@@ -124,6 +124,11 @@

124124 heartbeat_interval_secs: self.config().ci.heartbeat_interval_secs,

125125 attempt: job.attempt,

126126 },

127 trace: run

128 .event_payload

129 .get("_gitdata_trace")

130 .and_then(|value| value.as_str())

131 .map(str::to_owned),

127132 actions: Some(actions),

128133 })

129134 }

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

@@ -498,6 +498,11 @@

498498 heartbeat_interval_secs: self.config().ci.heartbeat_interval_secs,

499499 attempt: job.attempt,

500500 },

501 trace: run

502 .event_payload

503 .get("_gitdata_trace")

504 .and_then(|value| value.as_str())

505 .map(str::to_owned),

501506 actions: None,

502507 };

503508 Ok(payload)

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

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

6868 "new_oid": update.new_oid,

6969 "actor_id": event.actor_id,

7070 "event_id": event.id,

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

7172 }),

7273 commit_sha: update.new_oid.clone(),

7374 git_ref: update.name.clone(),

@@ -95,6 +96,7 @@

9596 "actor_id": event.actor_id,

9697 "event_id": event.id,

9798 "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}},

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

98100 }),

99101 commit_sha: event.head_sha.clone(),

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

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

@@ -16,6 +16,8 @@

1616 /// 输出语言（BCP-47 或 `auto`）；`None` 时用 `[ai].output_language`。

1717 #[serde(default)]

1818 language: Option<String>,

19 #[serde(default)]

20 trace: Option<String>,

1921 },

2022 CommitReview {

2123 job_id: Uuid,

@@ -23,6 +25,8 @@

2325 commit_sha: String,

2426 #[serde(default)]

2527 language: Option<String>,

28 #[serde(default)]

29 trace: Option<String>,

2630 },

2731 RepoSummary {

2832 job_id: Uuid,

@@ -32,6 +36,8 @@

3236 limit: Option<u64>,

3337 #[serde(default)]

3438 language: Option<String>,

39 #[serde(default)]

40 trace: Option<String>,

3541 },

3642 ReleaseNotes {

3743 job_id: Uuid,

@@ -42,6 +48,8 @@

4248 to_sha: String,

4349 #[serde(default)]

4450 language: Option<String>,

51 #[serde(default)]

52 trace: Option<String>,

4553 },

4654}

4755

@@ -116,4 +124,14 @@

116124 pub fn decode(payload: &[u8]) -> Result<Self, serde_json::Error> {

117125 serde_json::from_slice(payload)

118126 }

127

128 #[must_use]

129 pub fn trace(&self) -> Option<&str> {

130 match self {

131 Self::PullReview { trace, .. }

132 | Self::CommitReview { trace, .. }

133 | Self::RepoSummary { trace, .. }

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

135 }

136 }

119137}

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

@@ -72,6 +72,9 @@

7272 pub space_id: Uuid,

7373 pub actor_id: Option<Uuid>,

7474 pub refs: Vec<RefUpdate>,

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

76 #[serde(default)]

77 pub trace: Option<String>,

7578}

7679

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

@@ -88,6 +91,8 @@

8891 pub base_ref: String,

8992 #[serde(default)]

9093 pub head_repo_id: Option<Uuid>,

94 #[serde(default)]

95 pub trace: Option<String>,

9196}

9297

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

@@ -98,4 +103,17 @@

98103 pub actor_id: Option<Uuid>,

99104 pub event: String,

100105 pub payload: serde_json::Value,

106 #[serde(default)]

107 pub trace: Option<String>,

108}

109

110impl CiEvent {

111 #[must_use]

112 pub fn trace(&self) -> Option<&str> {

113 match self {

114 Self::Push(event) => event.trace.as_deref(),

115 Self::PullRequest(event) => event.trace.as_deref(),

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

117 }

118 }

101119}

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

@@ -135,6 +135,10 @@

135135 pub artifacts: Vec<ArtifactSpecDto>,

136136 pub needs_artifacts: Vec<NeedsArtifactDto>,

137137 pub lease: LeaseDto,

138 /// Trace context of the CI event that produced the run; lets the runner

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

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

141 pub trace: Option<String>,

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

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

140144}

Modified lib/email/src/events.rs +3 −0 View lib/email/src/events.rs

@@ -55,6 +55,7 @@

5555 pub attachments: Vec<Attachment>,

5656 pub metadata: HashMap<String, String>,

5757 pub queued_at_ms: i64,

58 pub trace: Option<String>,

5859}

5960

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

@@ -78,6 +79,7 @@

7879 pub recipients: Vec<BatchRecipient>,

7980 pub metadata: HashMap<String, String>,

8081 pub queued_at_ms: i64,

82 pub trace: Option<String>,

8183}

8284

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

@@ -94,6 +96,7 @@

9496 pub locale: Option<String>,

9597 pub metadata: HashMap<String, String>,

9698 pub queued_at_ms: i64,

99 pub trace: Option<String>,

97100}

98101

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

Modified lib/email/src/lib.rs +1 −0 View lib/email/src/lib.rs

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

55pub mod subscriber;

66pub mod template;

77pub mod templates;

8pub mod trace;

89pub mod transport;

910

1011pub mod runtime;

Modified lib/email/src/service.rs +3 −0 View lib/email/src/service.rs

@@ -126,6 +126,7 @@

126126 attachments: vec![],

127127 metadata: Default::default(),

128128 queued_at_ms: now_ms(),

129 trace: crate::trace::capture(),

129130 };

130131 let _ = self.send(req).await?;

131132 Ok(())

@@ -174,6 +175,7 @@

174175 attachments: vec![],

175176 metadata: meta,

176177 queued_at_ms: now_ms(),

178 trace: crate::trace::capture(),

177179 };

178180 match self.send(send_req).await {

179181 Ok(sent) => sent_list.push(sent),

@@ -251,6 +253,7 @@

251253 attachments: vec![],

252254 metadata: req.metadata,

253255 queued_at_ms: now_ms(),

256 trace: crate::trace::capture(),

254257 };

255258 self.send(send_req).await

256259 }

Modified lib/email/src/subscriber.rs +10 −1 View lib/email/src/subscriber.rs

@@ -1,6 +1,6 @@

11use std::time::Duration;

22

3use tracing::{debug, info};

3use tracing::{Instrument as _, debug, info};

44

55use crate::error::EmailError;

66use crate::events::{

@@ -72,6 +72,8 @@

7272 let svc = service.clone();

7373 move |req: SendEmailRequest, _meta: TypedMessageMeta| {

7474 let svc = svc.clone();

75 let span = tracing::info_span!("email.send", request_id = %req.request_id);

76 crate::trace::adopt(&span, req.trace.as_deref());

7577 async move {

7678 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

7779 let request_id = req.request_id;

@@ -113,6 +115,7 @@

113115 }

114116 Ok(())

115117 }

118 .instrument(span)

116119 }

117120 })

118121 .await?;

@@ -123,6 +126,8 @@

123126 let svc = service.clone();

124127 move |req: SendBatchEmailRequest, _meta: TypedMessageMeta| {

125128 let svc = svc.clone();

129 let span = tracing::info_span!("email.batch", request_id = %req.request_id);

130 crate::trace::adopt(&span, req.trace.as_deref());

126131 async move {

127132 let total = req.recipients.len();

128133 let request_id = req.request_id;

@@ -156,6 +161,7 @@

156161 }

157162 Ok(())

158163 }

164 .instrument(span)

159165 }

160166 })

161167 .await?;

@@ -166,6 +172,8 @@

166172 let svc = service.clone();

167173 move |req: RenderAndSendEmailRequest, _meta: TypedMessageMeta| {

168174 let svc = svc.clone();

175 let span = tracing::info_span!("email.render", request_id = %req.request_id);

176 crate::trace::adopt(&span, req.trace.as_deref());

169177 async move {

170178 let request_id = req.request_id;

171179 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

@@ -208,6 +216,7 @@

208216 }

209217 Ok(())

210218 }

219 .instrument(span)

211220 }

212221 })

213222 .await?;

Added lib/email/src/trace.rs +48 −0 View lib/email/src/trace.rs

@@ -0,0 +1,48 @@

1use std::collections::HashMap;

2

3use opentelemetry::Context;

4use opentelemetry::global;

5use opentelemetry::propagation::{Extractor, Injector};

6use tracing::Span;

7use tracing_opentelemetry::OpenTelemetrySpanExt;

8

9#[derive(Default)]

10struct Carrier(HashMap<String, String>);

11

12impl Injector for Carrier {

13 fn set(&mut self, key: &str, value: String) {

14 self.0.insert(key.to_string(), value);

15 }

16}

17

18impl Extractor for Carrier {

19 fn get(&self, key: &str) -> Option<&str> {

20 self.0.get(key).map(String::as_str)

21 }

22

23 fn keys(&self) -> Vec<&str> {

24 self.0.keys().map(String::as_str).collect()

25 }

26}

27

28pub fn capture() -> Option<String> {

29 let mut carrier = Carrier::default();

30 global::get_text_map_propagator(|propagator| {

31 propagator.inject_context(&Span::current().context(), &mut carrier);

32 });

33 carrier.0.get("traceparent").cloned()

34}

35

36pub fn adopt(span: &Span, traceparent: Option<&str>) {

37 let Some(traceparent) = traceparent else {

38 return;

39 };

40 let mut carrier = Carrier::default();

41 carrier

42 .0

43 .insert("traceparent".to_string(), traceparent.to_string());

44 let parent = global::get_text_map_propagator(|propagator| {

45 propagator.extract_with_context(&Context::new(), &carrier)

46 });

47 let _ = span.set_parent(parent);

48}

Modified lib/email/Cargo.toml +3 −1 View lib/email/Cargo.toml

@@ -8,13 +8,15 @@

88i18n = { workspace = true }

99lettre = { workspace = true }

1010maud = { workspace = true }

11opentelemetry = { workspace = true }

1112redis = { workspace = true }

1213serde = { workspace = true, features = ["derive"] }

1314serde_json = { workspace = true }

1415tokio = { workspace = true, features = ["signal", "rt-multi-thread", "macros"] }

1516tokio-util = { workspace = true }

17tracing = { workspace = true }

18tracing-opentelemetry = { workspace = true }

1619futures = { workspace = true }

17tracing = { workspace = true }

1820async-trait = { workspace = true }

1921uuid = { workspace = true, features = ["v7", "serde"] }

2022base64 = { workspace = true }

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

@@ -98,6 +98,8 @@

9898 /// 站内通知主键；仅用于日志关联，不参与投递决策。

9999 #[serde(default)]

100100 pub notification_id: Option<Uuid>,

101 #[serde(default)]

102 pub trace: Option<String>,

101103}

102104

103105impl PushJob {

@@ -130,6 +132,7 @@

130132 data,

131133 actor_id,

132134 notification_id,

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

133136 })

134137 }

135138

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

@@ -113,6 +113,7 @@

113113 crate::sync::ai_language::actor_language(self.db_read(), Some(creator)).await

114114 }

115115 },

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

116117 };

117118 let job_id =

118119 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

@@ -167,6 +167,7 @@

167167 Some(explicit) => Some(ci_protocol::ai_languages::normalize(&explicit)),

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

169169 },

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

170171 };

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

172173 .await

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

@@ -286,6 +286,7 @@

286286 from_sha: None,

287287 to_sha: release.sha.clone(),

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

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

289290 };

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

291292 .await

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

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

133133 repo_id,

134134 commit_sha: new.clone(),

135135 language: language.clone(),

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

136137 };

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

138139 }

@@ -149,6 +150,7 @@

149150 base_sha: None,

150151 limit: None,

151152 language: language.clone(),

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

152154 };

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

154156 }

@@ -243,6 +245,7 @@

243245 from_sha: previous,

244246 to_sha: target_sha.to_string(),

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

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

246249 };

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

248251 Ok(Some(job_id))

@@ -319,12 +322,14 @@

319322 pull_request_id: row.target_id?,

320323 head_sha: target_ref,

321324 language,

325 trace: None,

322326 }),

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

324328 job_id: row.id,

325329 repo_id: row.repo_id,

326330 commit_sha: target_ref,

327331 language,

332 trace: None,

328333 }),

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

330335 job_id: row.id,

@@ -333,6 +338,7 @@

333338 base_sha: None,

334339 limit: None,

335340 language,

341 trace: None,

336342 }),

337343 AiJobKind::ReleaseNotes => {

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

@@ -347,6 +353,7 @@

347353 from_sha: None,

348354 to_sha: release.sha,

349355 language,

356 trace: None,

350357 })

351358 }

352359 }

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

@@ -47,6 +47,8 @@

4747 pub channel_id: Uuid,

4848 pub actor_id: Option<Uuid>,

4949 pub refs: Vec<PostReceiveRefUpdate>,

50 #[serde(default)]

51 pub trace: Option<String>,

5052}

5153

5254impl PostReceiveTask {

@@ -63,6 +65,7 @@

6365 channel_id,

6466 actor_id,

6567 refs,

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

6669 }

6770 }

6871}

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

@@ -223,6 +223,7 @@

223223 space_id: channel_id,

224224 actor_id,

225225 refs: ci_refs,

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

226227 });

227228 match serde_json::to_vec(&event) {

228229 Ok(payload) => {

Modified lib/git/src/sync/worker.rs +10 −3 View lib/git/src/sync/worker.rs

@@ -9,7 +9,7 @@

99use sea_orm::EntityTrait;

1010use tokio::task::JoinHandle;

1111use tokio_util::sync::CancellationToken;

12use tracing::{debug, error, info, warn};

12use tracing::{Instrument as _, debug, error, info, warn};

1313

1414use crate::sync::error::{SyncError, SyncResult};

1515use crate::sync::event::PostReceiveTask;

@@ -167,6 +167,13 @@

167167 refs = task.refs.len(),

168168 "sync task received"

169169 );

170 let span = tracing::info_span!(

171 "git.sync",

172 task_id = %task.id,

173 repo_id = %task.repo_id,

174 refs = task.refs.len(),

175 );

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

170177

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

172179 Ok(Some(r)) => r,

@@ -200,7 +207,7 @@

200207 settings.clone(),

201208 );

202209

203 match rsync.sync().await {

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

204211 Ok(()) => {

205212 info!(

206213 consumer = id,

@@ -221,7 +228,7 @@

221228 "pull request refresh failed"

222229 );

223230 }

224 if let Err(e) = webhook.dispatch(&task).await {

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

225232 warn!(

226233 consumer = id,

227234 task_id = %task.id,

Modified lib/pulls/src/services/ai.rs +1 −0 View lib/pulls/src/services/ai.rs

@@ -78,6 +78,7 @@

7878 Some(explicit) => Some(ci_protocol::ai_languages::normalize(&explicit)),

7979 None => git::sync::actor_language(self.db_read(), actor).await,

8080 },

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

8182 };

8283 let job_id =

8384 git::sync::ai_task::submit_ai_task(self.db_write(), &self.state.redis, cfg, &task)

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

@@ -142,6 +142,7 @@

142142 base_sha: pull.base_sha.clone(),

143143 base_ref: pull.base_branch.clone(),

144144 head_repo_id: pull.head_repo_id,

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

145146 });

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

147148 Ok(payload) => payload,

@@ -199,6 +200,7 @@

199200 head_sha: pull.head_sha.clone(),

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

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

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

202204 };

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

204206 .await;

Modified lib/push/src/worker.rs +4 −1 View lib/push/src/worker.rs

@@ -14,6 +14,7 @@

1414use models::users::user_preferences;

1515use models::users::user_push_subscriptions::{self, PushSubscriptionPlatform};

1616use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};

17use tracing::Instrument as _;

1718use uuid::Uuid;

1819

1920use crate::client::{DeliveryOutcome, NotificationPayload, SubscriptionKeys, VapidKeys, deliver};

@@ -82,7 +83,9 @@

8283 }

8384 };

8485

85 if let Err(error) = self.handle(&job).await {

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

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

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

8689 tracing::warn!(

8790 %error,

8891 user = %job.user_id,

Modified web/src/shared/api/index.ts +7 −1 View web/src/shared/api/index.ts

@@ -2,12 +2,16 @@

22import fetchAdapter from 'alova/fetch';

33import { createApis, withConfigType, mountApis } from './createApis';

44import { recordApiResult } from './health';

5import { newTraceparent } from './trace';

56

67export const alovaInstance = createAlova({

78 baseURL: '/api/v1',

89 requestAdapter: fetchAdapter(),

910 beforeRequest: method => {

10 void method;

11 method.config.headers = {

12 ...method.config.headers,

13 traceparent: newTraceparent()

14 };

1115 },

1216 responded: {

1317 // Every settled response is reported to the health window before its body

@@ -87,6 +91,8 @@

8791 type ApiResult

8892} from './health';

8993

94export { newTraceparent } from './trace';

95

9096// Re-export the generated DTO types through the segment's public API so

9197// consumers never deep-import globals.d (regenerated file, see alova.config).

9298export type * from './globals.d'

Added web/src/shared/api/trace.ts +12 −0 View web/src/shared/api/trace.ts

@@ -0,0 +1,12 @@

1const TRACE_SAMPLE_RATE = 0.1

2

3function randomHex(bytes: number): string {

4 const buffer = new Uint8Array(bytes)

5 crypto.getRandomValues(buffer)

6 return Array.from(buffer, value => value.toString(16).padStart(2, '0')).join('')

7}

8

9export function newTraceparent(): string {

10 const sampled = Math.random() < TRACE_SAMPLE_RATE ? '01' : '00'

11 return `00-${randomHex(16)}-${randomHex(8)}-${sampled}`

12}

Modified web/src/shared/ws/client.ts +9 −1 View web/src/shared/ws/client.ts

@@ -14,6 +14,7 @@

1414} from './protocol'

1515import { requestWsJwt, isWsJwtError, type WsJwt } from './jwt'

1616import { TypingRegistry } from './typing'

17import { newTraceparent } from '@/shared/api'

1718

1819/**

1920 * Browser client for the chat WebSocket (`GET /ws/v1`, served by the im gateway).

@@ -380,7 +381,7 @@

380381 // The upgrade response re-offers the same protocol name (ws/mod.rs) and

381382 // JWTs only contain base64url / '.' characters, so they are valid

382383 // subprotocol tokens.

383 ws = new WebSocket(this.url, [`${JWT_PROTOCOL_PREFIX}${jwt.jwt}`])

384 ws = new WebSocket(withTraceparent(this.url), [`${JWT_PROTOCOL_PREFIX}${jwt.jwt}`])

384385 } catch (error) {

385386 if (generation !== this.generation) return

386387 this.registerFailure('handshake', error)

@@ -778,3 +779,10 @@

778779 const proto = window.location.protocol === 'https:' ? 'wss' : 'ws'

779780 return `${proto}://${window.location.host}${DEFAULT_PATH}`

780781}

782

783/** WebSocket handshakes cannot carry custom headers; the traceparent travels

784 * in the query string and is adopted by the im gateway. */

785function withTraceparent(url: string): string {

786 const separator = url.includes('?') ? '&' : '?'

787 return `${url}${separator}traceparent=${newTraceparent()}`

788}

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

@@ -2341,6 +2341,7 @@

23412341 "i18n",

23422342 "lettre",

23432343 "maud",

2344 "opentelemetry",

23442345 "redis",

23452346 "serde",

23462347 "serde_json",

@@ -2348,6 +2349,7 @@

23482349 "tokio",

23492350 "tokio-util",

23502351 "tracing",

2352 "tracing-opentelemetry",

23512353 "uuid",

23522354]

23532355

@@ -7944,6 +7946,7 @@

79447946 "ci-protocol",

79457947 "config",

79467948 "dotenvy",

7949 "export",

79477950 "figment",

79487951 "globset",

79497952 "reqwest 0.12.28",