From 9344ced3869b8fae790d16cdd3ba2ddc8cf2d11a Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Thu, 7 May 2026 20:42:57 -0500 Subject: [PATCH 1/2] Add GCP auth and Pulumi stack config to CI preview jobs The infra-preview and services-preview jobs were missing the GCP authentication and pulumi config steps that the deploy workflows already use, so pulumi preview couldn't authenticate to GCP and the services stack was missing its required githubToken config. Mirror the same setup the deploy jobs use (deploy-infra.yml for infra and build-and-deploy.yml's deploy job for services) so PR previews succeed for same-repo PRs. Co-Authored-By: Claude Opus 4.7 (1M context) --- .github/workflows/ci.yml | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7ca8d24..0116226 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,6 +41,16 @@ jobs: node-version: '20' cache: 'npm' + - uses: google-github-actions/auth@v2 + with: + workload_identity_provider: ${{ vars.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ vars.GCP_SERVICE_ACCOUNT_EMAIL }} + + - uses: google-github-actions/setup-gcloud@v2 + + - name: Install gke-gcloud-auth-plugin + run: gcloud components install gke-gcloud-auth-plugin + - name: Install dependencies run: npm ci @@ -50,6 +60,13 @@ jobs: requested-token-type: urn:pulumi:token-type:access_token:personal scope: user:${{ vars.PULUMI_USER }} + - name: Configure Pulumi stack + working-directory: infra + run: | + pulumi stack select --create ${{ vars.PULUMI_ORG }}/dev + pulumi config set gcp:project ${{ vars.GCP_PROJECT_ID }} + pulumi config set gcp:region ${{ vars.GCP_REGION }} + - uses: pulumi/actions@v5 with: command: preview @@ -70,6 +87,15 @@ jobs: node-version: '20' cache: 'npm' + - uses: google-github-actions/auth@v2 + with: + workload_identity_provider: ${{ vars.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ vars.GCP_SERVICE_ACCOUNT_EMAIL }} + + - uses: google-github-actions/setup-gcloud@v2 + with: + install_components: gke-gcloud-auth-plugin + - name: Install dependencies run: npm ci @@ -79,6 +105,19 @@ jobs: requested-token-type: urn:pulumi:token-type:access_token:personal scope: user:${{ vars.PULUMI_USER }} + - name: Configure Pulumi stack + working-directory: services + run: | + pulumi stack select --create ${{ vars.PULUMI_ORG }}/dev + pulumi config set gcp:project ${{ vars.GCP_PROJECT_ID }} + pulumi config set gcp:region ${{ vars.GCP_REGION }} + pulumi config set --secret githubToken '${{ secrets.GH_CONTROLLER_TOKEN }}' + if [ -n "${{ secrets.LOGFIRE_TOKEN }}" ]; then + pulumi config set --secret logfireToken '${{ secrets.LOGFIRE_TOKEN }}' + fi + pulumi config set imageTag ${{ github.sha }} + pulumi config set runnerRepoUrl '${{ github.server_url }}/${{ github.repository }}' + - uses: pulumi/actions@v5 with: command: preview From c7e90cdf452321e440f087d7e2ae33eb44915767 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Thu, 7 May 2026 20:33:37 -0500 Subject: [PATCH 2/2] Update the trigger comment in place instead of posting new comments Each benchmark job now owns one markdown "section" of the trigger comment, delimited by HTML comment markers. The controller serializes fetch+merge+PATCH per comment_id with an in-memory mutex so sibling jobs that share a trigger don't race when each writes its own section. Machine info (instance, uname, lscpu) collapses under a single
block to keep the running state quiet. Co-Authored-By: Claude Opus 4.7 (1M context) --- controller/migrations/004_section_body.sql | 5 + controller/src/bin/controller.rs | 7 + controller/src/bin/runner.rs | 15 +-- controller/src/comment_render.rs | 127 +++++++++++++++++ controller/src/db.rs | 90 ++++++++++++- controller/src/github.rs | 49 +++++++ controller/src/health.rs | 116 ++++++++++++++-- controller/src/job_manager.rs | 69 ++++++---- controller/src/lib.rs | 1 + controller/src/runner/bench_arrow.rs | 148 +++++++------------- controller/src/runner/bench_criterion.rs | 150 +++++++-------------- controller/src/runner/bench_standard.rs | 119 ++++++++-------- controller/src/runner/config.rs | 14 +- controller/src/runner/controller_client.rs | 8 +- controller/src/runner/poster.rs | 52 +++++-- 15 files changed, 638 insertions(+), 332 deletions(-) create mode 100644 controller/migrations/004_section_body.sql create mode 100644 controller/src/comment_render.rs diff --git a/controller/migrations/004_section_body.sql b/controller/migrations/004_section_body.sql new file mode 100644 index 0000000..f2c7e3f --- /dev/null +++ b/controller/migrations/004_section_body.sql @@ -0,0 +1,5 @@ +-- Per-job markdown section body, written by the runner each time the job's +-- visible state on the trigger comment changes (running → completed/failed). +-- The controller renders the trigger comment by concatenating section_body +-- across all jobs that share a comment_id. +ALTER TABLE benchmark_jobs ADD COLUMN section_body TEXT; diff --git a/controller/src/bin/controller.rs b/controller/src/bin/controller.rs index c89fc99..05553aa 100644 --- a/controller/src/bin/controller.rs +++ b/controller/src/bin/controller.rs @@ -34,12 +34,18 @@ async fn main() -> Result<()> { let token = CancellationToken::new(); let ready = health::ready_flag(); + let comment_locks = health::comment_locks(); + let server_config = health::ServerConfig { + runner_repo_url: config.runner_repo_url.clone(), + }; let health_handle = tokio::spawn(health::serve( token.clone(), ready.clone(), pool.clone(), gh.clone(), + comment_locks.clone(), + server_config, )); let poller = tokio::spawn(github_poller::poll_loop( @@ -53,6 +59,7 @@ async fn main() -> Result<()> { config.clone(), pool.clone(), gh.clone(), + comment_locks.clone(), token.clone(), )); diff --git a/controller/src/bin/runner.rs b/controller/src/bin/runner.rs index cf09fb2..e9ca7d1 100644 --- a/controller/src/bin/runner.rs +++ b/controller/src/bin/runner.rs @@ -6,7 +6,6 @@ use anyhow::Result; use tracing::{error, info}; -use benchmark_controller::github; use benchmark_controller::runner::config::{BenchType, RunnerConfig}; use benchmark_controller::runner::poster::CommentPoster; use benchmark_controller::runner::{bench_arrow, bench_criterion, bench_standard, shell}; @@ -67,27 +66,25 @@ async fn run_benchmark(config: &RunnerConfig, poster: &CommentPoster) -> Result< async fn post_error_comment(config: &RunnerConfig, poster: &CommentPoster) { let tail = shell::tail_log(20).await; - let footer = github::issues_footer(config.runner_repo_url.as_deref()); let body = format!( - "Benchmark for [this request]({}) failed.\n\n\ + "\u{1f6a8} **Benchmark failed**\n\n\ Last 20 lines of output:\n\
Click to expand\n\n\ ```\n\ {tail}\n\ ```\n\n\ -
{footer}", - config.comment_url, +
" ); - let pr_number = match config.pr_number() { + let comment_id = match config.comment_id_i64() { Ok(n) => n, Err(e) => { - error!(error = %e, "cannot post error comment: failed to parse PR number"); + error!(error = %e, "cannot post error section: failed to parse comment id"); return; } }; - if let Err(e) = poster.post_comment(&config.repo, pr_number, &body).await { - error!(error = %e, "failed to post error comment"); + if let Err(e) = poster.update_section(&config.repo, comment_id, &body).await { + error!(error = %e, "failed to post error section"); } } diff --git a/controller/src/comment_render.rs b/controller/src/comment_render.rs new file mode 100644 index 0000000..2e6c8b3 --- /dev/null +++ b/controller/src/comment_render.rs @@ -0,0 +1,127 @@ +//! Render the trigger comment body from per-job benchmark sections. +//! +//! Each runner job contributes one markdown "section" that describes its +//! current state (running, completed, failed). Sibling jobs that share a +//! trigger comment get their sections concatenated into a single block, +//! delimited by HTML comment markers so we can re-extract the user's +//! original body on subsequent updates. + +pub const MARKER_BEGIN: &str = ""; +pub const MARKER_END: &str = ""; + +/// Strip a previously-appended bot block from `body`, returning whatever the +/// user originally wrote. If the markers aren't present (first edit, or +/// human-edited comment), the entire body is treated as user content. +pub fn extract_user_body(body: &str) -> &str { + match body.find(MARKER_BEGIN) { + Some(idx) => body[..idx].trim_end_matches(['\n', ' ']), + None => body.trim_end_matches(['\n', ' ']), + } +} + +/// Build the full trigger-comment body: the user's original text, followed by +/// the bot block containing one section per job. Returns `original_body` +/// unchanged when there are no sections to render. +pub fn render(original_body: &str, sections: &[String], footer: &str) -> String { + let user = extract_user_body(original_body); + if sections.iter().all(|s| s.trim().is_empty()) { + return user.to_string(); + } + let mut out = String::with_capacity(user.len() + 256); + if !user.is_empty() { + out.push_str(user); + out.push_str("\n\n"); + } + out.push_str(MARKER_BEGIN); + out.push('\n'); + for (i, section) in sections.iter().enumerate() { + let trimmed = section.trim(); + if trimmed.is_empty() { + continue; + } + if i > 0 { + out.push_str("\n\n"); + } + out.push_str(trimmed); + } + if !footer.trim().is_empty() { + out.push_str("\n\n"); + out.push_str(footer.trim_start_matches('\n')); + } + out.push('\n'); + out.push_str(MARKER_END); + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_returns_original_when_no_marker() { + assert_eq!( + extract_user_body("run benchmark tpch"), + "run benchmark tpch" + ); + } + + #[test] + fn extract_strips_bot_block() { + let body = format!("run benchmark tpch\n\n{MARKER_BEGIN}\nstuff\n{MARKER_END}",); + assert_eq!(extract_user_body(&body), "run benchmark tpch"); + } + + #[test] + fn extract_handles_marker_at_start() { + let body = format!("{MARKER_BEGIN}\nstuff\n{MARKER_END}"); + assert_eq!(extract_user_body(&body), ""); + } + + #[test] + fn render_with_no_sections_returns_user_body() { + assert_eq!(render("hello", &[], ""), "hello"); + assert_eq!(render("hello", &["".to_string()], ""), "hello"); + } + + #[test] + fn render_appends_single_section() { + let out = render("run benchmark tpch", &["section A".to_string()], ""); + assert!(out.starts_with("run benchmark tpch\n\n")); + assert!(out.contains(MARKER_BEGIN)); + assert!(out.contains("section A")); + assert!(out.trim_end().ends_with(MARKER_END)); + } + + #[test] + fn render_multiple_sections_separated() { + let out = render( + "trigger", + &["A".to_string(), "B".to_string(), "C".to_string()], + "", + ); + let a = out.find('A').unwrap(); + let b = out.find('B').unwrap(); + let c = out.find('C').unwrap(); + assert!(a < b && b < c); + } + + #[test] + fn render_replaces_prior_bot_block() { + let prior = format!("trigger\n\n{MARKER_BEGIN}\nold section\n{MARKER_END}",); + let out = render(&prior, &["new section".to_string()], ""); + assert!(out.contains("new section")); + assert!(!out.contains("old section")); + // Only one marker pair + assert_eq!(out.matches(MARKER_BEGIN).count(), 1); + assert_eq!(out.matches(MARKER_END).count(), 1); + } + + #[test] + fn render_includes_footer_inside_block() { + let out = render("trigger", &["section".to_string()], "\n\n---\n[issue]"); + let begin = out.find(MARKER_BEGIN).unwrap(); + let end = out.find(MARKER_END).unwrap(); + let issue = out.find("[issue]").unwrap(); + assert!(begin < issue && issue < end); + } +} diff --git a/controller/src/db.rs b/controller/src/db.rs index 19e8d53..9e6bfdc 100644 --- a/controller/src/db.rs +++ b/controller/src/db.rs @@ -128,14 +128,15 @@ pub async fn set_runner_token(pool: &SqlitePool, job_id: i64, token: &str) -> Re Ok(()) } -/// Look up a running job by id and return its stored runner token, repo, and -/// PR number. Returns `None` if the job doesn't exist. +/// Look up a running job by id and return its stored runner token, repo, PR +/// number, and the trigger comment id. Returns `None` if the job doesn't exist. pub async fn get_job_for_comment( pool: &SqlitePool, job_id: i64, -) -> Result)>> { - let row = sqlx::query_as::<_, (String, i64, String, Option)>( - "SELECT repo, pr_number, status, runner_token FROM benchmark_jobs WHERE id = ?", +) -> Result, i64)>> { + let row = sqlx::query_as::<_, (String, i64, String, Option, i64)>( + "SELECT repo, pr_number, status, runner_token, comment_id \ + FROM benchmark_jobs WHERE id = ?", ) .bind(job_id) .fetch_optional(pool) @@ -143,6 +144,36 @@ pub async fn get_job_for_comment( Ok(row) } +/// Persist the markdown section a runner wants displayed on the trigger +/// comment for this job. Replaces any prior section body for the same job. +pub async fn set_section_body(pool: &SqlitePool, job_id: i64, body: &str) -> Result<()> { + sqlx::query( + "UPDATE benchmark_jobs SET section_body = ?, updated_at = datetime('now') WHERE id = ?", + ) + .bind(body) + .bind(job_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Return all (job_id, section_body) pairs for jobs sharing a trigger comment, +/// in insertion order, skipping jobs whose section hasn't been set yet. +pub async fn get_sections_for_comment( + pool: &SqlitePool, + comment_id: i64, +) -> Result> { + let rows = sqlx::query_as::<_, (i64, String)>( + "SELECT id, section_body FROM benchmark_jobs \ + WHERE comment_id = ? AND section_body IS NOT NULL \ + ORDER BY id", + ) + .bind(comment_id) + .fetch_all(pool) + .await?; + Ok(rows) +} + /// Return all jobs with status `running`. pub async fn get_active_jobs(pool: &SqlitePool) -> Result> { let jobs = sqlx::query_as::<_, BenchmarkJob>( @@ -412,16 +443,63 @@ mod tests { set_runner_token(&pool, id, "secret-abc").await.unwrap(); - let (repo, pr, status, token) = get_job_for_comment(&pool, id).await.unwrap().unwrap(); + let (repo, pr, status, token, comment_id) = + get_job_for_comment(&pool, id).await.unwrap().unwrap(); assert_eq!(repo, "apache/datafusion"); assert_eq!(pr, 42); assert_eq!(status, "pending"); assert_eq!(token.as_deref(), Some("secret-abc")); + assert_eq!(comment_id, 2000); // Missing job assert!(get_job_for_comment(&pool, 99_999).await.unwrap().is_none()); } + // ── section_body helpers ──────────────────────────────────────── + + #[tokio::test] + async fn section_body_round_trip_and_aggregate() { + let pool = test_pool().await; + mark_comment_seen(&pool, 3000, "apache/datafusion", 42, "alice", "2024-01-01") + .await + .unwrap(); + let id_a = insert_job(&pool, &test_job(3000)).await.unwrap(); + let id_b = insert_job(&pool, &test_job(3000)).await.unwrap(); + + // Initially neither job has a section body + assert!(get_sections_for_comment(&pool, 3000) + .await + .unwrap() + .is_empty()); + + set_section_body(&pool, id_a, "running tpch").await.unwrap(); + let sections = get_sections_for_comment(&pool, 3000).await.unwrap(); + assert_eq!(sections, vec![(id_a, "running tpch".to_string())]); + + set_section_body(&pool, id_b, "running clickbench") + .await + .unwrap(); + // Replace job A's section body + set_section_body(&pool, id_a, "completed tpch") + .await + .unwrap(); + + let sections = get_sections_for_comment(&pool, 3000).await.unwrap(); + assert_eq!( + sections, + vec![ + (id_a, "completed tpch".to_string()), + (id_b, "running clickbench".to_string()), + ] + ); + + // A different comment_id should not see these sections + assert!(get_sections_for_comment(&pool, 9999) + .await + .unwrap() + .is_empty()); + } + // ── update_job_status + get_active_jobs ───────────────────────── #[tokio::test] diff --git a/controller/src/github.rs b/controller/src/github.rs index 89716d2..27db372 100644 --- a/controller/src/github.rs +++ b/controller/src/github.rs @@ -164,6 +164,29 @@ impl GitHubClient { .await } + /// Send a PATCH request with retry logic. Returns the successful response. + async fn patch_with_retry(&self, url: &str, body: serde_json::Value) -> Result { + let url = url.to_string(); + + (|| { + let url = url.clone(); + let body = body.clone(); + async move { + let resp = self + .request_builder(self.client.patch(&url)) + .json(&body) + .send() + .await + .context("send request")?; + Self::check_response(resp, "PATCH").await + } + }) + .retry(ExponentialBuilder::default().with_max_times(3)) + .sleep(tokio::time::sleep) + .when(is_retryable) + .await + } + /// Fetch issue/PR comments updated since `since` (ISO 8601), paginating through all results. /// Caps at MAX_PAGES pages (10,000 comments). #[tracing::instrument(skip(self, since))] @@ -246,6 +269,32 @@ impl GitHubClient { Ok(pull.head.ref_) } + /// Fetch the body of an issue/PR comment. + #[tracing::instrument(skip(self))] + pub async fn get_comment_body(&self, repo: &str, comment_id: i64) -> Result { + #[derive(serde::Deserialize)] + struct Comment { + body: Option, + } + let url = format!("{API_BASE}/repos/{repo}/issues/comments/{comment_id}"); + let resp = self + .get_with_retry(&url, &[]) + .await + .context("fetch comment")?; + let comment: Comment = resp.json().await.context("parse comment json")?; + Ok(comment.body.unwrap_or_default()) + } + + /// Replace the body of an issue/PR comment. + #[tracing::instrument(skip(self, body))] + pub async fn update_comment(&self, repo: &str, comment_id: i64, body: &str) -> Result<()> { + let url = format!("{API_BASE}/repos/{repo}/issues/comments/{comment_id}"); + self.patch_with_retry(&url, serde_json::json!({ "body": body })) + .await + .context("update comment")?; + Ok(()) + } + /// Add a reaction (e.g. "rocket") to a comment. Logs a warning on failure instead of erroring. pub async fn post_reaction(&self, repo: &str, comment_id: i64, content: &str) -> Result<()> { let url = format!("{API_BASE}/repos/{repo}/issues/comments/{comment_id}/reactions"); diff --git a/controller/src/health.rs b/controller/src/health.rs index 3a9ec6a..9d5f29b 100644 --- a/controller/src/health.rs +++ b/controller/src/health.rs @@ -1,23 +1,27 @@ //! Minimal TCP-based HTTP server. //! //! Serves Kubernetes liveness/readiness probes plus a narrow -//! `POST /jobs/{id}/comment` endpoint used by benchmark runner pods to post -//! PR comments. The runner has no GitHub credentials of its own; it hands a -//! markdown body to the controller (authenticated with a per-job token -//! injected at pod-creation time) and the controller posts it via -//! [`GitHubClient::post_comment`]. - +//! `POST /jobs/{id}/comment` endpoint used by benchmark runner pods to update +//! the trigger comment. The runner has no GitHub credentials of its own; it +//! hands a markdown section body to the controller (authenticated with a +//! per-job token injected at pod-creation time) and the controller PATCHes +//! the trigger comment with the merged sections of all sibling jobs sharing +//! that comment. + +use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use sqlx::SqlitePool; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; use tracing::info; +use crate::comment_render; use crate::db; -use crate::github::GitHubClient; +use crate::github::{self, GitHubClient}; const OK: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"; const NOT_FOUND: &[u8] = b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n"; @@ -41,9 +45,40 @@ pub fn ready_flag() -> ReadyFlag { Arc::new(AtomicBool::new(false)) } +/// Per-trigger-comment async lock map. Serializes fetch+merge+PATCH for +/// sibling jobs that share a trigger comment so they don't race when each +/// posts its own section. +pub type CommentLocks = Arc>>>>; + +pub fn comment_locks() -> CommentLocks { + Arc::new(Mutex::new(HashMap::new())) +} + +async fn lock_for_comment(locks: &CommentLocks, comment_id: i64) -> Arc> { + let mut map = locks.lock().await; + map.entry(comment_id) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone() +} + +/// Configuration that the section-update handler needs beyond the per-request +/// state: the runner-repo footer URL is invariant for the lifetime of the +/// process and shared across all sections. +#[derive(Clone)] +pub struct ServerConfig { + pub runner_repo_url: Option, +} + /// Listen on `0.0.0.0:8080` and serve `/healthz`, `/readyz`, and /// `POST /jobs/{id}/comment`. -pub async fn serve(token: CancellationToken, ready: ReadyFlag, pool: SqlitePool, gh: GitHubClient) { +pub async fn serve( + token: CancellationToken, + ready: ReadyFlag, + pool: SqlitePool, + gh: GitHubClient, + locks: CommentLocks, + server_config: ServerConfig, +) { let listener = TcpListener::bind("0.0.0.0:8080") .await .expect("failed to bind health port 8080"); @@ -64,15 +99,24 @@ pub async fn serve(token: CancellationToken, ready: ReadyFlag, pool: SqlitePool, let ready = ready.clone(); let pool = pool.clone(); let gh = gh.clone(); - tokio::spawn(handle_conn(stream, ready, pool, gh)); + let locks = locks.clone(); + let server_config = server_config.clone(); + tokio::spawn(handle_conn(stream, ready, pool, gh, locks, server_config)); } info!("http server stopped"); } -async fn handle_conn(mut stream: TcpStream, ready: ReadyFlag, pool: SqlitePool, gh: GitHubClient) { +async fn handle_conn( + mut stream: TcpStream, + ready: ReadyFlag, + pool: SqlitePool, + gh: GitHubClient, + locks: CommentLocks, + server_config: ServerConfig, +) { let response: Vec = match read_request(&mut stream).await { - Ok(Some(req)) => match route(&req, &ready, &pool, &gh).await { + Ok(Some(req)) => match route(&req, &ready, &pool, &gh, &locks, &server_config).await { Ok(bytes) => bytes, Err(e) => { tracing::warn!(error = %e, "handler error"); @@ -96,6 +140,8 @@ async fn route( ready: &ReadyFlag, pool: &SqlitePool, gh: &GitHubClient, + locks: &CommentLocks, + server_config: &ServerConfig, ) -> anyhow::Result> { // Liveness if req.method == "GET" && req.path == "/healthz" { @@ -112,7 +158,7 @@ async fn route( // POST /jobs/{id}/comment if req.method == "POST" { if let Some(job_id) = parse_job_comment_path(&req.path) { - return handle_job_comment(req, job_id, pool, gh).await; + return handle_job_comment(req, job_id, pool, gh, locks, server_config).await; } } Ok(NOT_FOUND.to_vec()) @@ -123,6 +169,8 @@ async fn handle_job_comment( job_id: i64, pool: &SqlitePool, gh: &GitHubClient, + locks: &CommentLocks, + server_config: &ServerConfig, ) -> anyhow::Result> { // Auth: `Authorization: Bearer ` must match the row's runner_token. let bearer = match req.header("authorization") { @@ -141,7 +189,7 @@ async fn handle_job_comment( Some(r) => r, None => return Ok(NOT_FOUND.to_vec()), }; - let (repo, pr_number, status, stored_token) = row; + let (repo, _pr_number, status, stored_token, comment_id) = row; match stored_token { Some(tok) if constant_time_eq(tok.as_bytes(), supplied.as_bytes()) => {} _ => return Ok(UNAUTHORIZED.to_vec()), @@ -170,10 +218,50 @@ async fn handle_job_comment( return Ok(BAD_REQUEST.to_vec()); } - gh.post_comment(&repo, pr_number, &payload.body).await?; + // Serialize fetch+merge+PATCH per trigger comment so sibling jobs don't + // race. The lock is only held by this controller — direct-mode runners + // (main-tracking) do their own update without going through here. + update_trigger_comment( + pool, + gh, + locks, + server_config.runner_repo_url.as_deref(), + job_id, + &repo, + comment_id, + &payload.body, + ) + .await?; Ok(OK.to_vec()) } +/// Set this job's section body, then re-render and PATCH the trigger comment. +/// Holds the per-comment lock for the entire fetch+merge+PATCH so sibling +/// runners' updates don't interleave. +#[allow(clippy::too_many_arguments)] +pub async fn update_trigger_comment( + pool: &SqlitePool, + gh: &GitHubClient, + locks: &CommentLocks, + runner_repo_url: Option<&str>, + job_id: i64, + repo: &str, + comment_id: i64, + section_body: &str, +) -> anyhow::Result<()> { + let lock = lock_for_comment(locks, comment_id).await; + let _guard = lock.lock().await; + + db::set_section_body(pool, job_id, section_body).await?; + let sections = db::get_sections_for_comment(pool, comment_id).await?; + let section_bodies: Vec = sections.into_iter().map(|(_, body)| body).collect(); + let original = gh.get_comment_body(repo, comment_id).await?; + let footer = github::issues_footer(runner_repo_url); + let new_body = comment_render::render(&original, §ion_bodies, &footer); + gh.update_comment(repo, comment_id, &new_body).await?; + Ok(()) +} + fn parse_job_comment_path(path: &str) -> Option { let rest = path.strip_prefix("/jobs/")?; let (id, tail) = rest.split_once('/')?; diff --git a/controller/src/job_manager.rs b/controller/src/job_manager.rs index 2007353..63d8e3c 100644 --- a/controller/src/job_manager.rs +++ b/controller/src/job_manager.rs @@ -21,7 +21,8 @@ use tracing::{info, warn}; use crate::config::Config; use crate::db; -use crate::github::{self, GitHubClient}; +use crate::github::GitHubClient; +use crate::health::{self, CommentLocks}; use crate::models::{BenchmarkJob, JobStatus}; /// Infinite reconciliation loop that drives jobs through their lifecycle. @@ -47,6 +48,7 @@ pub async fn reconcile_loop( config: Config, pool: SqlitePool, gh: GitHubClient, + comment_locks: CommentLocks, token: tokio_util::sync::CancellationToken, ) -> Result<()> { let interval = tokio::time::Duration::from_secs(config.reconcile_interval_secs); @@ -56,10 +58,10 @@ pub async fn reconcile_loop( .context("failed to create kube client")?; loop { - if let Err(e) = reconcile_pending(&config, &pool, &gh, &kube_client).await { + if let Err(e) = reconcile_pending(&config, &pool, &gh, &comment_locks, &kube_client).await { warn!(error = %e, "reconcile pending error"); } - if let Err(e) = reconcile_active(&config, &pool, &gh, &kube_client).await { + if let Err(e) = reconcile_active(&config, &pool, &gh, &comment_locks, &kube_client).await { warn!(error = %e, "reconcile active error"); } tokio::select! { @@ -79,6 +81,7 @@ async fn reconcile_pending( config: &Config, pool: &SqlitePool, gh: &GitHubClient, + locks: &CommentLocks, kube: &KubeClient, ) -> Result<()> { let pending = db::get_pending_jobs(pool).await?; @@ -123,13 +126,20 @@ async fn reconcile_pending( ) .await?; - let comment_url = format!("{}#issuecomment-{}", job.pr_url, job.comment_id); - let footer = github::issues_footer(config.runner_repo_url.as_deref()); - let msg = format!( - "Failed to start benchmark for [this request]({comment_url}): {e}{footer}" - ); - if let Err(e) = gh.post_comment(&job.repo, job.pr_number, &msg).await { - warn!(error = %e, "failed to post error comment"); + let section = format!("\u{1f6a8} **Failed to start benchmark**\n\n```\n{e}\n```"); + if let Err(e) = health::update_trigger_comment( + pool, + gh, + locks, + config.runner_repo_url.as_deref(), + job.id, + &job.repo, + job.comment_id, + §ion, + ) + .await + { + warn!(error = %e, "failed to update trigger comment with start error"); } } } @@ -144,6 +154,7 @@ async fn reconcile_active( config: &Config, pool: &SqlitePool, gh: &GitHubClient, + locks: &CommentLocks, kube: &KubeClient, ) -> Result<()> { let active = db::get_active_jobs(pool).await?; @@ -197,7 +208,8 @@ async fn reconcile_active( // fire. The controller posts the notification here so the // PR doesn't go silent. if reason == "DeadlineExceeded" { - post_deadline_exceeded_comment(config, gh, &job, message).await; + post_deadline_exceeded_comment(config, gh, locks, pool, &job, message) + .await; } } } @@ -224,23 +236,21 @@ async fn reconcile_active( Ok(()) } -/// Post a PR comment when a benchmark Job hits `activeDeadlineSeconds`. -/// -/// The runner's own `post_error_comment` can't fire in this case because the -/// deadline SIGKILLs the pod. Failures to post are logged but not propagated — -/// the DB is already marked failed, and repeatedly retrying would risk double -/// comments. +/// Update the trigger comment when a benchmark Job hits +/// `activeDeadlineSeconds`. The runner's own error path can't fire because +/// the deadline SIGKILLs the pod. Failures are logged but not propagated — +/// the DB is already marked failed, and retrying could thrash the comment. async fn post_deadline_exceeded_comment( config: &Config, gh: &GitHubClient, + locks: &CommentLocks, + pool: &SqlitePool, job: &BenchmarkJob, k8s_message: &str, ) { let benchmarks = serde_json::from_str::>(&job.benchmarks) .map(|v| v.join(", ")) .unwrap_or_else(|_| job.benchmarks.clone()); - let comment_url = format!("{}#issuecomment-{}", job.pr_url, job.comment_id); - let footer = github::issues_footer(config.runner_repo_url.as_deref()); let k8s_detail = if k8s_message.is_empty() { String::new() } else { @@ -248,16 +258,27 @@ async fn post_deadline_exceeded_comment( "\n\n
Kubernetes message\n\n```\n{k8s_message}\n```\n\n
" ) }; - let body = format!( - "Benchmark for [this request]({comment_url}) hit the {deadline}s job deadline before finishing.\n\n\ - Benchmarks requested: `{benchmarks}`{k8s_detail}{footer}", + let section = format!( + "\u{23F1}\u{FE0F} **Benchmark hit the {deadline}s job deadline**\n\n\ + Benchmarks requested: `{benchmarks}`{k8s_detail}", deadline = config.active_deadline_secs, ); - if let Err(e) = gh.post_comment(&job.repo, job.pr_number, &body).await { + if let Err(e) = health::update_trigger_comment( + pool, + gh, + locks, + config.runner_repo_url.as_deref(), + job.id, + &job.repo, + job.comment_id, + §ion, + ) + .await + { warn!( comment_id = job.comment_id, error = %e, - "failed to post deadline-exceeded comment" + "failed to update trigger comment for deadline-exceeded" ); } } diff --git a/controller/src/lib.rs b/controller/src/lib.rs index 7ab29c6..fd15501 100644 --- a/controller/src/lib.rs +++ b/controller/src/lib.rs @@ -1,4 +1,5 @@ pub mod benchmarks; +pub mod comment_render; pub mod config; pub mod db; pub mod github; diff --git a/controller/src/runner/bench_arrow.rs b/controller/src/runner/bench_arrow.rs index 0c11a39..cb410e6 100644 --- a/controller/src/runner/bench_arrow.rs +++ b/controller/src/runner/bench_arrow.rs @@ -5,7 +5,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; use tracing::{info, warn}; -use crate::github; +use crate::runner::bench_standard::format_machine_details; use crate::runner::config::RunnerConfig; use crate::runner::git; use crate::runner::monitor; @@ -59,7 +59,7 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { // Pre-install stable toolchain to avoid rustup race in parallel builds git::rustup_stable().await?; - // Post "running" comment + // Post "running" section let uname = shell::uname().await; let instance_type = shell::node_instance_type().await; let pod_resources = shell::pod_resources(); @@ -75,27 +75,20 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { } else { format!("{} (merge-base)", &base_sha[..7.min(base_sha.len())]) }; - let footer = github::issues_footer(config.runner_repo_url.as_deref()); - let running_body = format!( - "\u{1f916} Arrow criterion benchmark running (GKE) | [trigger]({})\n\ - **Instance:** `{instance_type}` ({pod_resources}) | `{uname}`\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + let machine_details = format_machine_details(&instance_type, &pod_resources, &uname, &lscpu); + let running_section = format!( + "\u{1f916} **Arrow criterion benchmark running (GKE)**\n\n\ Comparing {changed_display} ({changed_sha}) to {baseline_label} \ [diff](https://github.com/{repo}/compare/{base_sha}..{changed_sha})\n\ BENCH_NAME={bench_name}\n\ BENCH_COMMAND={bench_command_display}\n\ - BENCH_FILTER={bench_filter}\n\ - Results will be posted here when complete{footer}", - config.comment_url, + BENCH_FILTER={bench_filter}\n\n\ + {machine_details}", repo = config.repo, ); - let pr_number = config.pr_number()?; + let comment_id = config.comment_id_i64()?; poster - .post_comment(&config.repo, pr_number, &running_body) + .update_section(&config.repo, comment_id, &running_section) .await?; // Compile both in parallel @@ -178,7 +171,7 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { }; // Compare and post results - let result_body = if baseline_available { + let result_section = if baseline_available { // Copy baselines into one target dir for critcmp copy_criterion_baselines(&base_dir, &branch_dir).await; @@ -191,15 +184,7 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { monitor::format_resource_comment("base (merge-base)", &base_stats.unwrap()), monitor::format_resource_comment("branch", &branch_stats), ); - format_result_comment( - &config.comment_url, - &report, - &resource_section, - &instance_type, - &pod_resources, - &lscpu, - &footer, - ) + format_result_section(&report, &resource_section, &machine_details) } else { let report = shell::run_command("critcmp", &[bench_branch_name.as_str()], &branch_dir) .await @@ -207,18 +192,10 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { let resource_section = monitor::format_resource_comment("branch", &branch_stats).to_string(); - format_branch_only_result_comment( - &config.comment_url, - &report, - &resource_section, - &instance_type, - &pod_resources, - &lscpu, - &footer, - ) + format_branch_only_result_section(&report, &resource_section, &machine_details) }; poster - .post_comment(&config.repo, pr_number, &result_body) + .update_section(&config.repo, comment_id, &result_section) .await?; Ok(()) @@ -252,24 +229,10 @@ async fn copy_criterion_baselines(base_dir: &Path, branch_dir: &Path) { } } -/// Format the result comment body. -fn format_result_comment( - comment_url: &str, - report: &str, - resource_section: &str, - instance_type: &str, - pod_resources: &str, - lscpu: &str, - footer: &str, -) -> String { +/// Format the per-job "completed" section for a baseline-comparison run. +fn format_result_section(report: &str, resource_section: &str, machine_details: &str) -> String { format!( - "\u{1f916} Arrow criterion benchmark completed (GKE) | [trigger]({comment_url})\n\n\ - **Instance:** `{instance_type}` ({pod_resources})\n\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + "\u{1f916} **Arrow criterion benchmark completed (GKE)**\n\n\
Details\n\

\n\n\ ```\n\ @@ -279,29 +242,19 @@ fn format_result_comment(

\n\n\
Resource Usage\n\n\ {resource_section}\ -
\n\ - {footer}" + \n\n\ + {machine_details}" ) } -/// Format the result comment body for branch-only runs (no baseline comparison). -fn format_branch_only_result_comment( - comment_url: &str, +/// Format the per-job "completed" section for branch-only runs. +fn format_branch_only_result_section( report: &str, resource_section: &str, - instance_type: &str, - pod_resources: &str, - lscpu: &str, - footer: &str, + machine_details: &str, ) -> String { format!( - "\u{1f916} Arrow criterion benchmark completed (GKE) | [trigger]({comment_url})\n\n\ - **Instance:** `{instance_type}` ({pod_resources})\n\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + "\u{1f916} **Arrow criterion benchmark completed (GKE)**\n\n\ **New benchmark — branch-only results (no baseline comparison)**\n\n\
Details\n\

\n\n\ @@ -312,8 +265,8 @@ fn format_branch_only_result_comment(

\n\n\
Resource Usage\n\n\ {resource_section}\ -
\n\ - {footer}" + \n\n\ + {machine_details}" ) } @@ -341,44 +294,41 @@ mod tests { } #[test] - fn result_comment_format() { - let comment = format_result_comment( - "https://example.com/comment", - "test report\n", - "resources\n", + fn result_section_format() { + let machine = format_machine_details( "c4a-standard-48", "12 vCPU / 65 GiB", + "uname", "lscpu output", - "", ); - assert!(comment.contains("Arrow criterion benchmark completed")); - assert!(comment.contains("[trigger](https://example.com/comment)")); - assert!(comment.contains("test report")); - assert!(comment.contains("Resource Usage")); - assert!(comment.contains("c4a-standard-48")); - assert!(comment.contains("12 vCPU / 65 GiB")); - assert!(comment.contains("lscpu output")); + let section = format_result_section("test report\n", "resources\n", &machine); + assert!(section.contains("Arrow criterion benchmark completed")); + assert!(section.contains("test report")); + assert!(section.contains("Resource Usage")); + assert!(section.contains("c4a-standard-48")); + assert!(section.contains("12 vCPU / 65 GiB")); + assert!(section.contains("lscpu output")); + assert!(!section.contains("[trigger]")); } #[test] - fn branch_only_result_comment_format() { - let comment = format_branch_only_result_comment( - "https://example.com/comment", - "branch report\n", - "branch resources\n", + fn branch_only_result_section_format() { + let machine = format_machine_details( "c4a-standard-48", "12 vCPU / 65 GiB", + "uname", "lscpu output", - "", ); - assert!(comment.contains("Arrow criterion benchmark completed")); - assert!(comment.contains("New benchmark — branch-only results")); - assert!(comment.contains("[trigger](https://example.com/comment)")); - assert!(comment.contains("branch report")); - assert!(comment.contains("Resource Usage")); - assert!(comment.contains("branch resources")); - assert!(comment.contains("c4a-standard-48")); - assert!(comment.contains("12 vCPU / 65 GiB")); - assert!(comment.contains("lscpu output")); + let section = + format_branch_only_result_section("branch report\n", "branch resources\n", &machine); + assert!(section.contains("Arrow criterion benchmark completed")); + assert!(section.contains("New benchmark — branch-only results")); + assert!(section.contains("branch report")); + assert!(section.contains("Resource Usage")); + assert!(section.contains("branch resources")); + assert!(section.contains("c4a-standard-48")); + assert!(section.contains("12 vCPU / 65 GiB")); + assert!(section.contains("lscpu output")); + assert!(!section.contains("[trigger]")); } } diff --git a/controller/src/runner/bench_criterion.rs b/controller/src/runner/bench_criterion.rs index 95397e1..114e8ac 100644 --- a/controller/src/runner/bench_criterion.rs +++ b/controller/src/runner/bench_criterion.rs @@ -5,7 +5,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; use tracing::{info, warn}; -use crate::github; +use crate::runner::bench_standard::format_machine_details; use crate::runner::config::RunnerConfig; use crate::runner::git; use crate::runner::monitor; @@ -58,7 +58,7 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { // Set up required benchmark data (e.g. sql_planner needs clickbench_partitioned) setup_benchmark_data(bench_name, &branch_dir, &base_dir).await; - // Post "running" comment + // Post "running" section let uname = shell::uname().await; let instance_type = shell::node_instance_type().await; let pod_resources = shell::pod_resources(); @@ -72,27 +72,20 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { } else { format!("{} (merge-base)", &base_sha[..7.min(base_sha.len())]) }; - let footer = github::issues_footer(config.runner_repo_url.as_deref()); - let running_body = format!( - "\u{1f916} Criterion benchmark running (GKE) | [trigger]({})\n\ - **Instance:** `{instance_type}` ({pod_resources}) | `{uname}`\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + let machine_details = format_machine_details(&instance_type, &pod_resources, &uname, &lscpu); + let running_section = format!( + "\u{1f916} **Criterion benchmark running (GKE)**\n\n\ Comparing {changed_display} ({changed_sha}) to {baseline_label} \ [diff](https://github.com/{repo}/compare/{base_sha}..{changed_sha})\n\ BENCH_NAME={bench_name}\n\ BENCH_COMMAND={bench_command_display}\n\ - BENCH_FILTER={bench_filter}\n\ - Results will be posted here when complete{footer}", - config.comment_url, + BENCH_FILTER={bench_filter}\n\n\ + {machine_details}", repo = config.repo, ); - let pr_number = config.pr_number()?; + let comment_id = config.comment_id_i64()?; poster - .post_comment(&config.repo, pr_number, &running_body) + .update_section(&config.repo, comment_id, &running_section) .await?; // Compile both in parallel @@ -175,7 +168,7 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { }; // Compare and post results - let result_body = if baseline_available { + let result_section = if baseline_available { // Copy baselines into one target dir for critcmp copy_criterion_baselines(&base_dir, &branch_dir).await; @@ -188,15 +181,7 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { monitor::format_resource_comment("base (merge-base)", &base_stats.unwrap()), monitor::format_resource_comment("branch", &branch_stats), ); - format_result_comment( - &config.comment_url, - &report, - &resource_section, - &instance_type, - &pod_resources, - &lscpu, - &footer, - ) + format_result_section(&report, &resource_section, &machine_details) } else { let report = shell::run_command("critcmp", &[bench_branch_name.as_str()], &branch_dir) .await @@ -204,18 +189,10 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { let resource_section = monitor::format_resource_comment("branch", &branch_stats).to_string(); - format_branch_only_result_comment( - &config.comment_url, - &report, - &resource_section, - &instance_type, - &pod_resources, - &lscpu, - &footer, - ) + format_branch_only_result_section(&report, &resource_section, &machine_details) }; poster - .post_comment(&config.repo, pr_number, &result_body) + .update_section(&config.repo, comment_id, &result_section) .await?; Ok(()) @@ -249,24 +226,10 @@ async fn copy_criterion_baselines(base_dir: &Path, branch_dir: &Path) { } } -/// Format the result comment body. -fn format_result_comment( - comment_url: &str, - report: &str, - resource_section: &str, - instance_type: &str, - pod_resources: &str, - lscpu: &str, - footer: &str, -) -> String { +/// Format the per-job "completed" section for a baseline-comparison run. +fn format_result_section(report: &str, resource_section: &str, machine_details: &str) -> String { format!( - "\u{1f916} Criterion benchmark completed (GKE) | [trigger]({comment_url})\n\n\ - **Instance:** `{instance_type}` ({pod_resources})\n\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + "\u{1f916} **Criterion benchmark completed (GKE)**\n\n\
Details\n\

\n\n\ ```\n\ @@ -276,29 +239,19 @@ fn format_result_comment(

\n\n\
Resource Usage\n\n\ {resource_section}\ -
\n\ - {footer}" + \n\n\ + {machine_details}" ) } -/// Format the result comment body for branch-only runs (no baseline comparison). -fn format_branch_only_result_comment( - comment_url: &str, +/// Format the per-job "completed" section for branch-only runs. +fn format_branch_only_result_section( report: &str, resource_section: &str, - instance_type: &str, - pod_resources: &str, - lscpu: &str, - footer: &str, + machine_details: &str, ) -> String { format!( - "\u{1f916} Criterion benchmark completed (GKE) | [trigger]({comment_url})\n\n\ - **Instance:** `{instance_type}` ({pod_resources})\n\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + "\u{1f916} **Criterion benchmark completed (GKE)**\n\n\ **New benchmark — branch-only results (no baseline comparison)**\n\n\
Details\n\

\n\n\ @@ -309,8 +262,8 @@ fn format_branch_only_result_comment(

\n\n\
Resource Usage\n\n\ {resource_section}\ -
\n\ - {footer}" + \n\n\ + {machine_details}" ) } @@ -397,45 +350,42 @@ mod tests { } #[test] - fn result_comment_format() { - let comment = format_result_comment( - "https://example.com/comment", - "test report\n", - "resources\n", + fn result_section_format() { + let machine = format_machine_details( "c4a-standard-48", "12 vCPU / 65 GiB", + "uname", "lscpu output", - "", ); - assert!(comment.contains("Criterion benchmark completed")); - assert!(comment.contains("[trigger](https://example.com/comment)")); - assert!(comment.contains("test report")); - assert!(comment.contains("
")); - assert!(comment.contains("Resource Usage")); - assert!(comment.contains("c4a-standard-48")); - assert!(comment.contains("12 vCPU / 65 GiB")); - assert!(comment.contains("lscpu output")); + let section = format_result_section("test report\n", "resources\n", &machine); + assert!(section.contains("Criterion benchmark completed")); + assert!(section.contains("test report")); + assert!(section.contains("
")); + assert!(section.contains("Resource Usage")); + assert!(section.contains("c4a-standard-48")); + assert!(section.contains("12 vCPU / 65 GiB")); + assert!(section.contains("lscpu output")); + assert!(!section.contains("[trigger]")); } #[test] - fn branch_only_result_comment_format() { - let comment = format_branch_only_result_comment( - "https://example.com/comment", - "branch report\n", - "branch resources\n", + fn branch_only_result_section_format() { + let machine = format_machine_details( "c4a-standard-48", "12 vCPU / 65 GiB", + "uname", "lscpu output", - "", ); - assert!(comment.contains("Criterion benchmark completed")); - assert!(comment.contains("New benchmark — branch-only results")); - assert!(comment.contains("[trigger](https://example.com/comment)")); - assert!(comment.contains("branch report")); - assert!(comment.contains("Resource Usage")); - assert!(comment.contains("branch resources")); - assert!(comment.contains("c4a-standard-48")); - assert!(comment.contains("12 vCPU / 65 GiB")); - assert!(comment.contains("lscpu output")); + let section = + format_branch_only_result_section("branch report\n", "branch resources\n", &machine); + assert!(section.contains("Criterion benchmark completed")); + assert!(section.contains("New benchmark — branch-only results")); + assert!(section.contains("branch report")); + assert!(section.contains("Resource Usage")); + assert!(section.contains("branch resources")); + assert!(section.contains("c4a-standard-48")); + assert!(section.contains("12 vCPU / 65 GiB")); + assert!(section.contains("lscpu output")); + assert!(!section.contains("[trigger]")); } } diff --git a/controller/src/runner/bench_standard.rs b/controller/src/runner/bench_standard.rs index d95aafb..92d181e 100644 --- a/controller/src/runner/bench_standard.rs +++ b/controller/src/runner/bench_standard.rs @@ -5,7 +5,6 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; use tracing::info; -use crate::github; use crate::runner::config::RunnerConfig; use crate::runner::git; use crate::runner::monitor::{self, ResourceStats}; @@ -71,12 +70,12 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { "/tmp/base_build.log", ); - // Post "running" comment + // Post "running" section let uname = shell::uname().await; let instance_type = shell::node_instance_type().await; let pod_resources = shell::pod_resources(); let lscpu = shell::lscpu().await; - let pr_number = config.pr_number()?; + let comment_id = config.comment_id_i64()?; // Resolve display names for the comparison let changed_display = config.changed_ref.as_deref().unwrap_or(&branch_name); @@ -88,24 +87,17 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { format!("{} (merge-base)", &base_sha[..7.min(base_sha.len())]) }; - let footer = github::issues_footer(config.runner_repo_url.as_deref()); - let running_body = format!( - "\u{1f916} Benchmark running (GKE) | [trigger]({})\n\ - **Instance:** `{instance_type}` ({pod_resources}) | `{uname}`\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + let machine_details = format_machine_details(&instance_type, &pod_resources, &uname, &lscpu); + let running_section = format!( + "\u{1f916} **Benchmark running (GKE)**\n\n\ Comparing {changed_display} ({changed_sha}) to {baseline_label} \ [diff](https://github.com/{repo}/compare/{base_sha}..{changed_sha}) \ - using: {benchmarks}\n\ - Results will be posted here when complete{footer}", - config.comment_url, + using: {benchmarks}\n\n\ + {machine_details}", repo = config.repo, ); poster - .post_comment(&config.repo, pr_number, &running_body) + .update_section(&config.repo, comment_id, &running_section) .await?; // Wait for builds @@ -197,22 +189,31 @@ pub async fn run(config: &RunnerConfig, poster: &CommentPoster) -> Result<()> { .context("bench.sh compare")?; let resource_section = format_resource_section(&base_stats_list, &branch_stats_list); - let result_body = format_result_comment( - &config.comment_url, - &report, - &resource_section, - &instance_type, - &pod_resources, - &lscpu, - &footer, - ); + let result_section = format_result_section(&report, &resource_section, &machine_details); poster - .post_comment(&config.repo, pr_number, &result_body) + .update_section(&config.repo, comment_id, &result_section) .await?; Ok(()) } +/// Single collapsed `
` block holding instance/pod/uname/lscpu so the +/// trigger comment stays compact while the run is still in progress. +pub(crate) fn format_machine_details( + instance_type: &str, + pod_resources: &str, + uname: &str, + lscpu: &str, +) -> String { + format!( + "
Machine info\n\n\ + **Instance:** `{instance_type}` ({pod_resources})\n\n\ + **uname:** `{uname}`\n\n\ + ```\n{lscpu}\n```\n\n\ +
" + ) +} + /// Run a single benchmark on one side (base or branch). /// /// For TPC-H variants we bypass `bench.sh run` and invoke the prebuilt `dfbench` @@ -416,24 +417,12 @@ fn format_resource_section( section } -/// Format the result comment body. -fn format_result_comment( - comment_url: &str, - report: &str, - resource_section: &str, - instance_type: &str, - pod_resources: &str, - lscpu: &str, - footer: &str, -) -> String { +/// Format the per-job "completed" section that gets merged into the trigger +/// comment. The section is bracketed by the controller-side renderer, so this +/// only contains the parts specific to a single benchmark run. +fn format_result_section(report: &str, resource_section: &str, machine_details: &str) -> String { format!( - "\u{1f916} Benchmark completed (GKE) | [trigger]({comment_url})\n\n\ - **Instance:** `{instance_type}` ({pod_resources})\n\n\ -
CPU Details (lscpu)\n\n\ - ```\n\ - {lscpu}\n\ - ```\n\n\ -
\n\n\ + "\u{1f916} **Benchmark completed (GKE)**\n\n\
Details\n\

\n\n\ ```\n\ @@ -443,8 +432,8 @@ fn format_result_comment(

\n\n\
Resource Usage\n\n\ {resource_section}\ -
\n\ - {footer}" +
\n\n\ + {machine_details}" ) } @@ -453,25 +442,35 @@ mod tests { use super::*; #[test] - fn result_comment_format() { - let comment = format_result_comment( - "https://example.com/comment", - "test report\n", - "resources\n", + fn result_section_format() { + let machine = format_machine_details( "c4a-standard-48", "12 vCPU / 65 GiB", + "Linux node-1", "lscpu output", - "", ); - assert!(comment.contains("Benchmark completed")); - assert!(comment.contains("[trigger](https://example.com/comment)")); - assert!(comment.contains("test report")); - assert!(comment.contains("
")); - assert!(comment.contains("Resource Usage")); - assert!(comment.contains("resources")); - assert!(comment.contains("c4a-standard-48")); - assert!(comment.contains("12 vCPU / 65 GiB")); - assert!(comment.contains("lscpu output")); + let section = format_result_section("test report\n", "resources\n", &machine); + assert!(section.contains("Benchmark completed")); + assert!(section.contains("test report")); + assert!(section.contains("
")); + assert!(section.contains("Resource Usage")); + assert!(section.contains("resources")); + assert!(section.contains("c4a-standard-48")); + assert!(section.contains("12 vCPU / 65 GiB")); + assert!(section.contains("Linux node-1")); + assert!(section.contains("lscpu output")); + // No trigger link — we now edit the trigger comment in place. + assert!(!section.contains("[trigger]")); + } + + #[test] + fn machine_details_collapsed() { + let m = format_machine_details("c4a-1", "1 vCPU / 1 GiB", "uname-foo", "lscpu-foo"); + assert!(m.starts_with("
")); + assert!(m.contains("Machine info")); + assert!(m.contains("c4a-1")); + assert!(m.contains("uname-foo")); + assert!(m.contains("lscpu-foo")); } #[test] diff --git a/controller/src/runner/config.rs b/controller/src/runner/config.rs index e6e2e56..0d364f6 100644 --- a/controller/src/runner/config.rs +++ b/controller/src/runner/config.rs @@ -114,9 +114,10 @@ impl RunnerConfig { /// Build the [`CommentPoster`] implied by [`Self::poster_mode`]. pub fn build_poster(&self) -> CommentPoster { match &self.poster_mode { - PosterMode::Direct { github_token } => { - CommentPoster::Direct(GitHubClient::new(github_token)) - } + PosterMode::Direct { github_token } => CommentPoster::Direct { + client: GitHubClient::new(github_token), + runner_repo_url: self.runner_repo_url.clone(), + }, PosterMode::Proxy { controller_url, job_id, @@ -129,6 +130,13 @@ impl RunnerConfig { } } + /// The trigger comment id, parsed from `COMMENT_ID`. + pub fn comment_id_i64(&self) -> Result { + self.comment_id + .parse() + .with_context(|| format!("invalid COMMENT_ID: {}", self.comment_id)) + } + /// The repo clone URL. pub fn repo_url(&self) -> String { format!("https://github.com/{}.git", self.repo) diff --git a/controller/src/runner/controller_client.rs b/controller/src/runner/controller_client.rs index 50ccff7..3df3375 100644 --- a/controller/src/runner/controller_client.rs +++ b/controller/src/runner/controller_client.rs @@ -28,12 +28,10 @@ impl ControllerClient { } } - /// Post a comment on the PR associated with this runner's job. `repo` and - /// `pr_number` are accepted for signature parity with - /// [`crate::github::GitHubClient::post_comment`] but are ignored — the - /// controller resolves both from the job's DB row. + /// Replace this job's section of the trigger comment. The controller + /// merges sibling-job sections and PATCHes the comment. #[tracing::instrument(skip(self, body), fields(job_id = %self.job_id))] - pub async fn post_comment(&self, _repo: &str, _pr_number: i64, body: &str) -> Result<()> { + pub async fn post_section(&self, body: &str) -> Result<()> { let url = format!("{}/jobs/{}/comment", self.base_url, self.job_id); let payload = json!({ "body": body }); diff --git a/controller/src/runner/poster.rs b/controller/src/runner/poster.rs index 8d0b931..01a063c 100644 --- a/controller/src/runner/poster.rs +++ b/controller/src/runner/poster.rs @@ -1,26 +1,54 @@ -//! `CommentPoster` chooses between posting PR comments directly to GitHub -//! (with a `GITHUB_TOKEN` — used by the scheduled main-tracking workflow) -//! and proxying through the controller (used by PR-triggered runs, which -//! have no GitHub creds in the pod). +//! `CommentPoster` chooses between editing the trigger comment directly on +//! GitHub (with a `GITHUB_TOKEN` — used by the scheduled main-tracking +//! workflow) and proxying through the controller (used by PR-triggered runs, +//! which have no GitHub creds in the pod). +//! +//! Each runner job owns a single "section" of the trigger comment (the part +//! that changes from "running" to "completed"/"failed"). Updating a section +//! re-renders the trigger comment in place rather than posting a new comment. -use anyhow::Result; +use anyhow::{Context, Result}; -use crate::github::GitHubClient; +use crate::comment_render; +use crate::github::{self, GitHubClient}; use crate::runner::controller_client::ControllerClient; #[derive(Clone)] pub enum CommentPoster { - /// Post directly to GitHub using a `GITHUB_TOKEN`. - Direct(GitHubClient), - /// Proxy through the controller's `POST /jobs/{id}/comment` endpoint. + /// Edit comments directly on GitHub using a `GITHUB_TOKEN`. Used by the + /// main-tracking workflow where one runner owns the entire trigger + /// comment, so no inter-job coordination is required. + Direct { + client: GitHubClient, + runner_repo_url: Option, + }, + /// Proxy through the controller's `POST /jobs/{id}/comment` endpoint — + /// the controller serializes sibling-job updates per trigger comment. Proxy(ControllerClient), } impl CommentPoster { - pub async fn post_comment(&self, repo: &str, pr_number: i64, body: &str) -> Result<()> { + /// Set the section this runner contributes to the trigger comment, + /// re-rendering the comment so the new section becomes visible. + /// Replaces any prior section the same runner posted. + pub async fn update_section(&self, repo: &str, comment_id: i64, section: &str) -> Result<()> { match self { - Self::Direct(c) => c.post_comment(repo, pr_number, body).await, - Self::Proxy(c) => c.post_comment(repo, pr_number, body).await, + Self::Direct { + client, + runner_repo_url, + } => { + let original = client + .get_comment_body(repo, comment_id) + .await + .context("fetch trigger comment")?; + let footer = github::issues_footer(runner_repo_url.as_deref()); + let body = comment_render::render(&original, &[section.to_string()], &footer); + client + .update_comment(repo, comment_id, &body) + .await + .context("update trigger comment") + } + Self::Proxy(c) => c.post_section(section).await, } } }