Send past request durations for individual collectors into the frontend

Author: Kobzol

database/src/pool.rs

@@ -277,6 +277,12 @@ pub trait Connection: Send + Sync {
         &self,
     ) -> anyhow::Result<HashMap<String, Vec<BenchmarkJob>>>;
 
+    /// Return all jobs associated with the given benchmark request tags.
+    async fn get_jobs_of_benchmark_requests(
+        &self,
+        tags: &[String],
+    ) -> anyhow::Result<Vec<BenchmarkJob>>;
+
     /// Get all of the configuration for all of the collectors
     async fn get_collector_configs(&self) -> anyhow::Result<Vec<CollectorConfig>>;
 

database/src/pool/postgres.rs

@@ -526,6 +526,7 @@ pub struct CachedStatements {
     get_last_n_completed_requests_with_errors: Statement,
     get_jobs_of_in_progress_benchmark_requests: Statement,
     load_pending_benchmark_requests: Statement,
+    get_jobs_of_benchmark_requests: Statement,
 }
 
 pub struct PostgresTransaction<'a> {
@@ -763,6 +764,11 @@ impl PostgresConnection {
                     FROM pending
                     LEFT JOIN benchmark_request as parent ON parent.tag = pending.parent_sha
                 ")).await.unwrap(),
+                get_jobs_of_benchmark_requests: conn.prepare("
+                    SELECT *
+                    FROM job_queue
+                    WHERE request_tag = ANY($1)
+                ").await.unwrap(),
             }),
             conn,
         }
@@ -773,6 +779,44 @@ impl PostgresConnection {
 const BENCHMARK_REQUEST_COLUMNS: &str =
     "tag, parent_sha, pr, commit_type, status, created_at, completed_at, backends, profiles, commit_date, duration_ms, targets";
 
+/// Parse a benchmark job out of a row.
+/// Expects to be used with `SELECT * FROM job_queue`.
+fn parse_benchmark_job_from_row(row: &Row) -> anyhow::Result<BenchmarkJob> {
+    let started_at = row.get::<_, Option<DateTime<Utc>>>(7);
+    let status = row.get::<_, &str>(9);
+    let collector_name = row.get::<_, Option<String>>(11);
+    let status = match status {
+        BENCHMARK_JOB_STATUS_QUEUED_STR => BenchmarkJobStatus::Queued,
+        BENCHMARK_JOB_STATUS_IN_PROGRESS_STR => BenchmarkJobStatus::InProgress {
+            started_at: started_at.expect("started_at was null for an in progress job"),
+            collector_name: collector_name.expect("Collector is missing for an in progress job"),
+        },
+        BENCHMARK_JOB_STATUS_FAILURE_STR | BENCHMARK_JOB_STATUS_SUCCESS_STR => {
+            BenchmarkJobStatus::Completed {
+                started_at: started_at.expect("started_at was null for a finished job"),
+                completed_at: row.get::<_, DateTime<Utc>>(8),
+                collector_name: collector_name
+                    .expect("Collector is missing for an in progress job"),
+                success: status == BENCHMARK_JOB_STATUS_SUCCESS_STR,
+            }
+        }
+        _ => panic!("Invalid job status {status}"),
+    };
+    Ok(BenchmarkJob {
+        id: row.get::<_, i32>(0) as u32,
+        request_tag: row.get::<_, String>(1),
+        target: Target::from_str(row.get::<_, &str>(2)).map_err(|e| anyhow::anyhow!(e))?,
+        backend: CodegenBackend::from_str(row.get::<_, &str>(3)).map_err(|e| anyhow::anyhow!(e))?,
+        profile: Profile::from_str(row.get::<_, &str>(4)).map_err(|e| anyhow::anyhow!(e))?,
+        benchmark_set: BenchmarkSet(row.get::<_, i32>(5) as u32),
+        created_at: row.get::<_, DateTime<Utc>>(6),
+        status,
+        deque_counter: row.get::<_, i32>(10) as u32,
+        kind: BenchmarkJobKind::from_str(row.get::<_, &str>(12)).map_err(|e| anyhow::anyhow!(e))?,
+        is_optional: row.get::<_, bool>(13),
+    })
+}
+
 #[async_trait::async_trait]
 impl<P> Connection for P
 where
@@ -2000,43 +2044,7 @@ where
 
         let mut request_to_jobs: HashMap<String, Vec<BenchmarkJob>> = HashMap::new();
         for row in rows {
-            let started_at = row.get::<_, Option<DateTime<Utc>>>(7);
-            let status = row.get::<_, &str>(9);
-            let collector_name = row.get::<_, Option<String>>(11);
-            let status = match status {
-                BENCHMARK_JOB_STATUS_QUEUED_STR => BenchmarkJobStatus::Queued,
-                BENCHMARK_JOB_STATUS_IN_PROGRESS_STR => BenchmarkJobStatus::InProgress {
-                    started_at: started_at.expect("started_at was null for an in progress job"),
-                    collector_name: collector_name
-                        .expect("Collector is missing for an in progress job"),
-                },
-                BENCHMARK_JOB_STATUS_FAILURE_STR | BENCHMARK_JOB_STATUS_SUCCESS_STR => {
-                    BenchmarkJobStatus::Completed {
-                        started_at: started_at.expect("started_at was null for a finished job"),
-                        completed_at: row.get::<_, DateTime<Utc>>(8),
-                        collector_name: collector_name
-                            .expect("Collector is missing for an in progress job"),
-                        success: status == BENCHMARK_JOB_STATUS_SUCCESS_STR,
-                    }
-                }
-                _ => panic!("Invalid job status {status}"),
-            };
-            let job = BenchmarkJob {
-                id: row.get::<_, i32>(0) as u32,
-                request_tag: row.get::<_, String>(1),
-                target: Target::from_str(row.get::<_, &str>(2)).map_err(|e| anyhow::anyhow!(e))?,
-                backend: CodegenBackend::from_str(row.get::<_, &str>(3))
-                    .map_err(|e| anyhow::anyhow!(e))?,
-                profile: Profile::from_str(row.get::<_, &str>(4))
-                    .map_err(|e| anyhow::anyhow!(e))?,
-                benchmark_set: BenchmarkSet(row.get::<_, i32>(5) as u32),
-                created_at: row.get::<_, DateTime<Utc>>(6),
-                status,
-                deque_counter: row.get::<_, i32>(10) as u32,
-                kind: BenchmarkJobKind::from_str(row.get::<_, &str>(12))
-                    .map_err(|e| anyhow::anyhow!(e))?,
-                is_optional: row.get::<_, bool>(13),
-            };
+            let job = parse_benchmark_job_from_row(&row)?;
             request_to_jobs
                 .entry(job.request_tag.clone())
                 .or_default()
@@ -2045,6 +2053,17 @@ where
         Ok(request_to_jobs)
     }
 
+    async fn get_jobs_of_benchmark_requests(
+        &self,
+        tags: &[String],
+    ) -> anyhow::Result<Vec<BenchmarkJob>> {
+        let rows = self
+            .conn()
+            .query(&self.statements().get_jobs_of_benchmark_requests, &[&tags])
+            .await?;
+        rows.iter().map(parse_benchmark_job_from_row).collect()
+    }
+
     async fn get_collector_configs(&self) -> anyhow::Result<Vec<CollectorConfig>> {
         let rows = self
             .conn()

database/src/pool/sqlite.rs

@@ -1193,6 +1193,13 @@ impl Connection for SqliteConnection {
         no_queue_implementation_abort!()
     }
 
+    async fn get_jobs_of_benchmark_requests(
+        &self,
+        _tags: &[String],
+    ) -> anyhow::Result<Vec<BenchmarkJob>> {
+        no_queue_implementation_abort!()
+    }
+
     async fn get_collector_configs(&self) -> anyhow::Result<Vec<CollectorConfig>> {
         no_queue_implementation_abort!()
     }

site/frontend/src/pages/status/data.ts

@@ -30,6 +30,11 @@ export type BenchmarkJob = {
   dequeCounter: number;
 };
 
+export type PastRequestDuration = {
+  requestTag: String;
+  job_duration_s: number;
+};
+
 export type CollectorConfig = {
   name: string;
   target: string;
@@ -39,6 +44,7 @@ export type CollectorConfig = {
   dateAdded: string;
   commitSha: string;
   jobs: BenchmarkJob[];
+  pastRequestDurations: PastRequestDuration[];
 };
 
 export type StatusResponse = {

site/src/api.rs

@@ -413,6 +413,14 @@ pub mod status {
         pub deque_counter: u32,
     }
 
+    #[derive(Serialize, Debug)]
+    #[serde(rename_all = "camelCase")]
+    pub struct PastRequestDuration {
+        pub request_tag: String,
+        /// Sum of (completed_at - started_at) for all jobs this collector ran for the request, in seconds.
+        pub job_duration_s: u64,
+    }
+
     #[derive(Serialize, Debug)]
     #[serde(rename_all = "camelCase")]
     pub struct Collector {
@@ -424,6 +432,7 @@ pub mod status {
         pub date_added: DateTime<Utc>,
         pub commit_sha: String,
         pub jobs: Vec<BenchmarkJob>,
+        pub past_request_durations: Vec<PastRequestDuration>,
     }
 
     #[derive(Serialize, Debug)]

site/src/request_handlers/status_page.rs

@@ -1,4 +1,5 @@
 use crate::api::status;
+use crate::api::status::PastRequestDuration;
 use crate::job_queue::build_queue;
 use crate::load::SiteCtxt;
 use chrono::{DateTime, Utc};
@@ -7,6 +8,7 @@ use database::{
     BenchmarkRequestType, Connection,
 };
 use hashbrown::HashMap;
+use std::cmp::Reverse;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -17,6 +19,12 @@ pub async fn handle_status_page(ctxt: Arc<SiteCtxt>) -> anyhow::Result<status::R
     let queue = build_queue(&*conn).await?;
     let completed = conn.get_last_n_completed_benchmark_requests(10).await?;
 
+    let completed_tags = completed
+        .iter()
+        .filter_map(|req| req.request.tag())
+        .map(|t| t.to_owned())
+        .collect::<Vec<String>>();
+
     // Figure out approximately how long was the most recent master benchmark request
     let expected_duration = completed
         .iter()
@@ -28,7 +36,11 @@ pub async fn handle_status_page(ctxt: Arc<SiteCtxt>) -> anyhow::Result<status::R
         .next()
         .unwrap_or(Duration::from_secs(3600));
 
-    let in_progress_jobs = conn.get_jobs_of_in_progress_benchmark_requests().await?;
+    let (in_progress_jobs, past_jobs) = tokio::join!(
+        conn.get_jobs_of_in_progress_benchmark_requests(),
+        conn.get_jobs_of_benchmark_requests(&completed_tags)
+    );
+    let (in_progress_jobs, past_jobs) = (in_progress_jobs?, past_jobs?);
 
     // Here we compute the estimated end time for queued requests, and convert the requests to their
     // frontend representation.
@@ -91,7 +103,7 @@ pub async fn handle_status_page(ctxt: Arc<SiteCtxt>) -> anyhow::Result<status::R
             .map(|req| request_to_ui(&req.request, req.errors, None)),
     );
 
-    let collectors = build_collectors(conn.as_ref(), &in_progress_jobs).await?;
+    let collectors = build_collectors(conn.as_ref(), &in_progress_jobs, &past_jobs).await?;
 
     Ok(status::Response {
         requests,
@@ -102,11 +114,56 @@ pub async fn handle_status_page(ctxt: Arc<SiteCtxt>) -> anyhow::Result<status::R
 async fn build_collectors(
     conn: &dyn Connection,
     in_progress_jobs: &HashMap<String, Vec<BenchmarkJob>>,
+    past_jobs: &[BenchmarkJob],
 ) -> anyhow::Result<Vec<status::Collector>> {
+    // Collector -> request tag -> sum of job durations
+    let mut collector_request_durations: HashMap<&str, HashMap<&str, Duration>> = HashMap::new();
+    let mut request_start_time: HashMap<&str, DateTime<Utc>> = HashMap::new();
+
+    for job in past_jobs {
+        let BenchmarkJobStatus::Completed {
+            started_at,
+            completed_at,
+            collector_name,
+            ..
+        } = job.status()
+        else {
+            continue;
+        };
+        let start_time = request_start_time.entry(job.request_tag()).or_default();
+        *start_time = *(&*start_time).min(started_at);
+
+        // Accumulate per-collector and per-request job duration
+        let job_duration = (*completed_at - *started_at).to_std().unwrap_or_default();
+        *collector_request_durations
+            .entry(collector_name.as_str())
+            .or_default()
+            .entry(job.request_tag())
+            .or_default() += job_duration;
+    }
+
     let collectors = conn.get_collector_configs().await?;
     let mut collector_map: HashMap<String, status::Collector> = collectors
         .into_iter()
         .map(|c| {
+            let mut past_request_durations = collector_request_durations
+                .remove(c.name())
+                .unwrap_or_default()
+                .into_iter()
+                .map(|(request, collector_duration)| PastRequestDuration {
+                    request_tag: request.to_owned(),
+                    job_duration_s: collector_duration.as_secs(),
+                })
+                .collect::<Vec<_>>();
+            // Sort from the most recent benchmark request
+            past_request_durations.sort_by_key(|request| {
+                Reverse(
+                    request_start_time
+                        .get(request.request_tag.as_str())
+                        .copied()
+                        .unwrap_or_default(),
+                )
+            });
             (
                 c.name().to_string(),
                 status::Collector {
@@ -121,6 +178,7 @@ async fn build_collectors(
                         .map(|s| s.to_owned())
                         .unwrap_or_else(|| "<unknown>".to_owned()),
                     jobs: vec![],
+                    past_request_durations,
                 },
             )
         })