Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion src/http/router.rs
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ impl<B> OnResponse<B> for HttpRequestOnResponse {
fn on_response(self, response: &Response<B>, _latency: Duration, span: &Span) {
span.record("http.response.status_code", response.status().as_u16());
if self.should_mark_error(response.status()) {
span.record("otel.status_code", "ERROR");
crate::telemetry::record_span_error(span);
span.record("error.type", response.status().as_u16().to_string());
}
}
Expand Down
2 changes: 1 addition & 1 deletion src/polling/git.rs
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ impl GitFetcher for MainGitFetcher {
) -> Result<CommitHash, CommitHashError> {
let result = get_latest_hash_inner(&self.repo, self.timeout, repo_url, branch).await;
if result.is_err() {
tracing::Span::current().record("otel.status_code", "ERROR");
crate::telemetry::mark_current_span_error();
}
result
}
Expand Down
4 changes: 2 additions & 2 deletions src/polling/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ async fn process_branch_updates(
) -> Result<(), PollingError> {
let result = process_branch_updates_inner(ctx, updated_branches).await;
if result.is_err() {
tracing::Span::current().record("otel.status_code", "ERROR");
crate::telemetry::mark_current_span_error();
}
result
}
Expand Down Expand Up @@ -172,7 +172,7 @@ async fn process_single_branch(
) -> Result<(), RepositoryError> {
let result = process_single_branch_inner(repo, branch_info, tx).await;
if result.is_err() {
tracing::Span::current().record("otel.status_code", "ERROR");
crate::telemetry::mark_current_span_error();
}
result
}
Expand Down
16 changes: 16 additions & 0 deletions src/telemetry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,22 @@ pub fn serialize_current_span_context() -> Option<String> {
.ok()
}

/// Records the OpenTelemetry `ERROR` status on `span`.
///
/// Only effective if `span` declares the `otel.status_code` field
/// (e.g. through `#[tracing::instrument]`); otherwise the record is dropped.
pub(crate) fn record_span_error(span: &tracing::Span) {
span.record("otel.status_code", "ERROR");
}

/// Records the OpenTelemetry `ERROR` status on the current span.
///
/// Only effective if the current span declares the `otel.status_code` field
/// (e.g. through `#[tracing::instrument]`); otherwise the record is dropped.
pub(crate) fn mark_current_span_error() {
record_span_error(&tracing::Span::current());
}

/// Adds a span link to the given tracing span
/// from a serialized OpenTelemetry span context string, if valid.
pub fn add_link_from_serialized_context(span: &tracing::Span, span_context: Option<&str>) {
Expand Down
31 changes: 16 additions & 15 deletions src/trigger/auth.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,10 @@ use tracing::info;
use crate::{
config::Config,
model::Subscription,
trigger::error::{AuthError, RequestError},
trigger::{
error::{AuthError, RequestError},
github::{apply_api_headers, endpoint},
},
};

/// Provides authentication functionality.
Expand Down Expand Up @@ -111,21 +114,19 @@ pub(super) async fn request_iat(
token: String,
}

let api_url = format!(
"{}/app/installations/{}/access_tokens",
config.github_api.base_url.as_str().trim_end_matches('/'),
sub.gh_app_installation_id
let api_url = endpoint(
&config.github_api,
&format!(
"/app/installations/{}/access_tokens",
sub.gh_app_installation_id
),
);
let response = http_client
.post(&api_url)
.bearer_auth(jwt)
.header("Accept", config.github_api.accept_header.to_string())
.header(
"X-GitHub-Api-Version",
config.github_api.version.to_string(),
)
.send()
.await?;
let response = apply_api_headers(
&config.github_api,
http_client.post(&api_url).bearer_auth(jwt),
)
.send()
.await?;

if response.status().is_success() {
let response_json = response.json::<IatResponse>().await?;
Expand Down
41 changes: 15 additions & 26 deletions src/trigger/dispatch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ use crate::{
trigger::{
TriggerEngine,
error::{RequestError, WorkflowTriggerError},
github::{apply_api_headers, endpoint},
},
};

Expand Down Expand Up @@ -84,16 +85,12 @@ async fn send_repository_dispatch(
trigger: &TriggerQueueItem,
sub_with_branch: &SubscriptionWithBranch,
) -> Result<(), WorkflowTriggerError> {
let api_url = format!(
"{}/repos/{}/dispatches",
engine
.ctx
.config
.github_api
.base_url
.as_str()
.trim_end_matches('/'),
sub_with_branch.subscription.target_repo
let api_url = endpoint(
&engine.ctx.config.github_api,
&format!(
"/repos/{}/dispatches",
sub_with_branch.subscription.target_repo
),
);

let payload = serde_json::json!({
Expand All @@ -114,21 +111,13 @@ async fn send_repository_dispatch(
payload
);

let response = engine
.http_client
.post(&api_url)
.bearer_auth(iat)
.header(
"Accept",
engine.ctx.config.github_api.accept_header.to_string(),
)
.header(
"X-GitHub-Api-Version",
engine.ctx.config.github_api.version.to_string(),
)
.json(&payload)
.send()
.await?;
let response = apply_api_headers(
&engine.ctx.config.github_api,
engine.http_client.post(&api_url).bearer_auth(iat),
)
.json(&payload)
.send()
.await?;

if response.status().is_success() {
info!(
Expand All @@ -141,7 +130,7 @@ async fn send_repository_dispatch(
Ok(())
} else {
let span = tracing::Span::current();
span.record("otel.status_code", "ERROR");
crate::telemetry::record_span_error(&span);
span.record("error.type", response.status().as_u16().to_string());
Err(WorkflowTriggerError::Api(RequestError::Response {
status: response.status(),
Expand Down
109 changes: 109 additions & 0 deletions src/trigger/github.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
//! Helpers to build GitHub API requests.

use reqwest::RequestBuilder;

use crate::config::GitHubApiConfig;

/// Returns the full API URL for `path` based on the configured base URL.
///
/// A leading `/` on `path` is optional: exactly one separator is inserted
/// between the base URL and the path.
pub(super) fn endpoint(config: &GitHubApiConfig, path: &str) -> String {
format!(
"{}/{}",
config.base_url.as_str().trim_end_matches('/'),
path.trim_start_matches('/')
)
}

/// Adds the GitHub API `Accept` and `X-GitHub-Api-Version` headers
/// to `request`.
pub(super) fn apply_api_headers(
config: &GitHubApiConfig,
request: RequestBuilder,
) -> RequestBuilder {
request
.header("Accept", config.accept_header.to_string())
.header("X-GitHub-Api-Version", config.version.to_string())
}

#[cfg(test)]
mod tests {
#![allow(
clippy::panic,
clippy::expect_used,
clippy::todo,
clippy::unimplemented,
clippy::indexing_slicing
)]

use url::Url;

use crate::config::GitHubApiConfig;
use crate::domain::{AcceptHeader, ApiVersion};

use super::{apply_api_headers, endpoint};

fn test_config() -> GitHubApiConfig {
GitHubApiConfig {
base_url: Url::parse("https://api.github.com").unwrap(),
version: ApiVersion::new("2026-03-10".to_string()).unwrap(),
accept_header: AcceptHeader::new("application/vnd.github+json".to_string()).unwrap(),
}
}

#[test]
fn endpoint_without_trailing_slash() {
let config = GitHubApiConfig {
base_url: Url::parse("https://api.github.com/api/v3").unwrap(),
..test_config()
};
assert_eq!(
endpoint(&config, "/repos/org/repo/dispatches"),
"https://api.github.com/api/v3/repos/org/repo/dispatches"
);
}

#[test]
fn endpoint_with_trailing_slash() {
let config = GitHubApiConfig {
base_url: Url::parse("https://api.github.com/").unwrap(),
..test_config()
};
assert_eq!(
endpoint(&config, "/app/installations/1/access_tokens"),
"https://api.github.com/app/installations/1/access_tokens"
);
}

#[test]
fn endpoint_without_leading_slash() {
let config = test_config();
assert_eq!(
endpoint(&config, "repos/org/repo/dispatches"),
"https://api.github.com/repos/org/repo/dispatches"
);
}

#[test]
fn apply_api_headers_sets_expected_headers() {
let config = test_config();
let client = reqwest::Client::new();
let request = apply_api_headers(&config, client.get("https://example.com"))
.build()
.unwrap();
assert_eq!(
request.headers().get("Accept").unwrap().to_str().unwrap(),
"application/vnd.github+json"
);
assert_eq!(
request
.headers()
.get("X-GitHub-Api-Version")
.unwrap()
.to_str()
.unwrap(),
"2026-03-10"
);
}
}
1 change: 1 addition & 0 deletions src/trigger/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ use crate::{
mod auth;
mod dispatch;
pub mod error;
mod github;
mod process;

pub use auth::{Authenticator, GitHubAuthenticator};
Expand Down
6 changes: 3 additions & 3 deletions src/trigger/process.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ pub(super) async fn process_trigger(
) -> Result<(), WorkflowTriggerError> {
let result = process_trigger_inner(engine, trigger).await;
if result.is_err() {
tracing::Span::current().record("otel.status_code", "ERROR");
crate::telemetry::mark_current_span_error();
}
result
}
Expand Down Expand Up @@ -70,7 +70,7 @@ async fn process_trigger_inner(
}
Err(e) => {
let span = tracing::Span::current();
span.record("otel.status_code", "ERROR");
crate::telemetry::record_span_error(&span);
span.record("error.type", "dispatch_failed");
warn!("Dispatch failed: {e}");
if let Err(retry_err) = schedule_retry(engine, trigger, e).await {
Expand Down Expand Up @@ -120,7 +120,7 @@ async fn schedule_retry(

if next_retry_count as u32 >= max_attempts {
let span = tracing::Span::current();
span.record("otel.status_code", "ERROR");
crate::telemetry::record_span_error(&span);
span.record("error.type", "retries_exhausted");
tracing::warn!(
"Task {} failed after {} attempts: {e}",
Expand Down
Loading