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