diff --git a/database/src/lib.rs b/database/src/lib.rs index f96df1184..9633044bc 100644 --- a/database/src/lib.rs +++ b/database/src/lib.rs @@ -957,10 +957,10 @@ impl BenchmarkRequest { } } - pub fn pr(&self) -> Option<&u32> { + pub fn pr(&self) -> Option { match &self.commit_type { BenchmarkRequestType::Try { pr, .. } | BenchmarkRequestType::Master { pr, .. } => { - Some(pr) + Some(*pr) } BenchmarkRequestType::Release { .. } => None, } @@ -986,6 +986,10 @@ impl BenchmarkRequest { self.commit_date } + pub fn commit_type(&self) -> &BenchmarkRequestType { + &self.commit_type + } + pub fn is_master(&self) -> bool { matches!(self.commit_type, BenchmarkRequestType::Master { .. }) } @@ -1248,12 +1252,9 @@ pub struct InProgressRequestWithJobs { pub parent: Option<(BenchmarkRequest, Vec)>, } -/// The data that can be retrived from the database directly to populate the -/// status page #[derive(Debug, PartialEq)] -pub struct PartialStatusPageData { - /// A Vector of; completed requests with any associated errors - pub completed_requests: Vec<(BenchmarkRequest, Vec)>, - /// In progress requests along with their associated jobs - pub in_progress: Vec, +pub struct BenchmarkRequestWithErrors { + pub request: BenchmarkRequest, + /// Benchmark (name) -> error + pub errors: HashMap, } diff --git a/database/src/pool.rs b/database/src/pool.rs index 1f5587407..571b4510e 100644 --- a/database/src/pool.rs +++ b/database/src/pool.rs @@ -1,8 +1,8 @@ use crate::selector::CompileTestCase; use crate::{ ArtifactCollection, ArtifactId, ArtifactIdNumber, BenchmarkJob, BenchmarkJobConclusion, - BenchmarkRequest, BenchmarkRequestIndex, BenchmarkRequestStatus, BenchmarkSet, CodegenBackend, - CollectorConfig, CompileBenchmark, PartialStatusPageData, Target, + BenchmarkRequest, BenchmarkRequestIndex, BenchmarkRequestStatus, BenchmarkRequestWithErrors, + BenchmarkSet, CodegenBackend, CollectorConfig, CompileBenchmark, Target, }; use crate::{CollectionId, Index, Profile, QueuedCommit, Scenario, Step}; use chrono::{DateTime, Utc}; @@ -214,7 +214,7 @@ pub trait Connection: Send + Sync { commit_date: DateTime, ) -> anyhow::Result<()>; - /// Add a benchmark job to the job queue. + /// Add a benchmark job to the job queue and return its ID. async fn enqueue_benchmark_job( &self, request_tag: &str, @@ -222,7 +222,7 @@ pub trait Connection: Send + Sync { backend: CodegenBackend, profile: Profile, benchmark_set: u32, - ) -> anyhow::Result<()>; + ) -> anyhow::Result; /// Returns a set of compile-time benchmark test cases that were already computed for the /// given artifact. @@ -265,7 +265,7 @@ pub trait Connection: Send + Sync { /// Try and mark the benchmark_request as completed. Will return `true` if /// it has been marked as completed else `false` meaning there was no change - async fn mark_benchmark_request_as_completed(&self, tag: &str) -> anyhow::Result; + async fn maybe_mark_benchmark_request_as_completed(&self, tag: &str) -> anyhow::Result; /// Mark the job as completed. Sets the status to 'failed' or 'success' /// depending on the enum's completed state being a success @@ -275,7 +275,20 @@ pub trait Connection: Send + Sync { conclusion: BenchmarkJobConclusion, ) -> anyhow::Result<()>; - async fn get_status_page_data(&self) -> anyhow::Result; + /// Return the last `count` completed benchmark requests, along with all errors associated with + /// them. + /// + /// The requests will be ordered from most recently to least recently completed. + async fn get_last_n_completed_benchmark_requests( + &self, + count: u64, + ) -> anyhow::Result>; + + /// Return jobs of all requests that are currently in progress, and the jobs of their parents. + /// The keys of the hashmap contain the request tags. + async fn get_jobs_of_in_progress_benchmark_requests( + &self, + ) -> anyhow::Result>>; /// Get all of the configuration for all of the collectors async fn get_collector_configs(&self) -> anyhow::Result>; @@ -403,8 +416,8 @@ impl Pool { mod tests { use super::*; use crate::metric::Metric; + use crate::tests::builder::{job, RequestBuilder}; use crate::tests::run_postgres_test; - use crate::BenchmarkJobStatus; use crate::{tests::run_db_test, BenchmarkRequestType, Commit, CommitType, Date}; use chrono::Utc; use std::str::FromStr; @@ -450,7 +463,7 @@ mod tests { .unwrap(); assert!(db - .mark_benchmark_request_as_completed(request_tag) + .maybe_mark_benchmark_request_as_completed(request_tag) .await .unwrap()); } @@ -461,8 +474,7 @@ mod tests { // This is essentially testing the database testing framework is // wired up correctly. Though makes sense that there should be // an empty vector returned if there are no pstats. - let db = ctx.db_client(); - let result = db.connection().await.get_pstats(&[], &[]).await; + let result = ctx.db().get_pstats(&[], &[]).await; let expected: Vec>> = vec![]; assert_eq!(result, expected); @@ -474,21 +486,19 @@ mod tests { #[tokio::test] async fn artifact_storage() { run_db_test(|ctx| async { - let db = ctx.db_client(); + let db = ctx.db(); let time = chrono::DateTime::from_str("2021-09-01T00:00:00.000Z").unwrap(); let artifact_one = ArtifactId::from(create_commit("abc", time, CommitType::Master)); let artifact_two = ArtifactId::Tag("nightly-2025-05-14".to_string()); - let artifact_one_id_number = db.connection().await.artifact_id(&artifact_one).await; - let artifact_two_id_number = db.connection().await.artifact_id(&artifact_two).await; + let artifact_one_id_number = db.artifact_id(&artifact_one).await; + let artifact_two_id_number = db.artifact_id(&artifact_two).await; // We cannot arbitrarily add random sizes to the artifact size // table, as there is a constraint that the artifact must actually // exist before attaching something to it. - let db = db.connection().await; - // Artifact one inserts db.record_artifact_size(artifact_one_id_number, "llvm.so", 32) .await; @@ -518,8 +528,8 @@ mod tests { #[tokio::test] async fn multiple_requests_same_sha() { run_postgres_test(|ctx| async { - let db = ctx.db_client(); - let db = db.connection().await; + let db = ctx.db(); + db.insert_benchmark_request(&BenchmarkRequest::create_master( "a-sha-1", "parent-sha-1", @@ -542,7 +552,7 @@ mod tests { #[tokio::test] async fn multiple_non_completed_try_requests() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let target = Target::X86_64UnknownLinuxGnu; let collector_name = "collector-1"; let benchmark_set = 1; @@ -565,8 +575,8 @@ mod tests { .await .unwrap(); - complete_request(&*db, "sha-parent-1", collector_name, benchmark_set, target).await; - complete_request(&*db, "sha1", collector_name, benchmark_set, target).await; + complete_request(db, "sha-parent-1", collector_name, benchmark_set, target).await; + complete_request(db, "sha1", collector_name, benchmark_set, target).await; // This should be fine, req_a was completed db.insert_benchmark_request(&req_b).await.unwrap(); @@ -584,8 +594,7 @@ mod tests { #[tokio::test] async fn multiple_master_requests_same_pr() { run_postgres_test(|ctx| async { - let db = ctx.db_client(); - let db = db.connection().await; + let db = ctx.db(); db.insert_benchmark_request(&BenchmarkRequest::create_master( "a-sha-1", @@ -613,7 +622,7 @@ mod tests { #[tokio::test] async fn load_pending_benchmark_requests() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let time = chrono::DateTime::from_str("2021-09-01T00:00:00.000Z").unwrap(); let target = Target::X86_64UnknownLinuxGnu; let collector_name = "collector-1"; @@ -638,7 +647,7 @@ mod tests { db.insert_benchmark_request(req).await.unwrap(); } - complete_request(&*db, "1.79.0", collector_name, benchmark_set, target).await; + complete_request(db, "1.79.0", collector_name, benchmark_set, target).await; db.update_benchmark_request_status("sha-2", BenchmarkRequestStatus::InProgress) .await @@ -659,8 +668,7 @@ mod tests { #[tokio::test] async fn attach_shas_to_try_benchmark_request() { run_postgres_test(|ctx| async { - let db = ctx.db_client(); - let db = db.connection().await; + let db = ctx.db(); let req = BenchmarkRequest::create_try_without_artifacts(42, "", ""); @@ -689,7 +697,7 @@ mod tests { assert_eq!(req_db.tag(), Some("sha1")); assert_eq!(req_db.parent_sha(), Some("sha-parent-1")); - assert_eq!(req_db.pr(), Some(&42)); + assert_eq!(req_db.pr(), Some(42)); Ok(ctx) }) @@ -699,8 +707,8 @@ mod tests { #[tokio::test] async fn enqueue_benchmark_job() { run_postgres_test(|ctx| async { - let db = ctx.db_client(); - let db = db.connection().await; + let db = ctx.db(); + let time = chrono::DateTime::from_str("2021-09-01T00:00:00.000Z").unwrap(); let benchmark_request = BenchmarkRequest::create_master("sha-1", "parent-sha-1", 42, time); @@ -730,7 +738,7 @@ mod tests { #[tokio::test] async fn get_compile_test_cases_with_data() { run_db_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let collection = db.collection_id("test").await; let artifact = db @@ -789,7 +797,7 @@ mod tests { #[tokio::test] async fn get_collector_config_error_if_not_exist() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let collector_config_result = db.start_collector("collector-1", "foo").await.unwrap(); @@ -803,7 +811,7 @@ mod tests { #[tokio::test] async fn add_collector_config() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let mut inserted_config = db .add_collector_config("collector-1", Target::X86_64UnknownLinuxGnu, 1, true) @@ -828,7 +836,7 @@ mod tests { #[tokio::test] async fn dequeue_benchmark_job_empty_queue() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let benchmark_job_result = db .dequeue_benchmark_job( @@ -849,7 +857,7 @@ mod tests { #[tokio::test] async fn dequeue_benchmark_job() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let time = chrono::DateTime::from_str("2021-09-01T00:00:00.000Z").unwrap(); let collector_config = db @@ -918,7 +926,7 @@ mod tests { #[tokio::test] async fn mark_request_as_complete_empty() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let time = chrono::DateTime::from_str("2021-09-01T00:00:00.000Z").unwrap(); let insert_result = db @@ -932,7 +940,7 @@ mod tests { .await .unwrap(); assert!(db - .mark_benchmark_request_as_completed("sha-1") + .maybe_mark_benchmark_request_as_completed("sha-1") .await .unwrap()); Ok(ctx) @@ -943,7 +951,7 @@ mod tests { #[tokio::test] async fn mark_request_as_complete() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; + let db = ctx.db(); let time = chrono::DateTime::from_str("2021-09-01T00:00:00.000Z").unwrap(); let benchmark_set = BenchmarkSet(0u32); let tag = "sha-1"; @@ -988,17 +996,18 @@ mod tests { .await .unwrap(); - db.mark_benchmark_request_as_completed(tag).await.unwrap(); + db.maybe_mark_benchmark_request_as_completed(tag) + .await + .unwrap(); /* From the status page view we can see that the duration has been * updated. Albeit that it will be a very short duration. */ - let status_page_view = db.get_status_page_data().await.unwrap(); - let req = &status_page_view - .completed_requests + let completed = db.get_last_n_completed_benchmark_requests(1).await.unwrap(); + let req = &completed .iter() - .find(|it| it.0.tag() == Some(tag)) + .find(|it| it.request.tag() == Some(tag)) .unwrap() - .0; + .request; assert!(matches!( req.status(), @@ -1018,136 +1027,110 @@ mod tests { } #[tokio::test] - async fn get_status_page_data() { + async fn get_collector_configs() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; - let benchmark_set = BenchmarkSet(0u32); - let time = chrono::DateTime::from_str("2021-09-01T00:00:00.000Z").unwrap(); - let tag = "sha-1"; - let tag_two = "sha-2"; - let collector_name = "collector-1"; + let db = ctx.db(); let target = Target::X86_64UnknownLinuxGnu; - db.add_collector_config(collector_name, target, benchmark_set.0, true) - .await - .unwrap(); - - let benchmark_request = BenchmarkRequest::create_release(tag, time); - db.insert_benchmark_request(&benchmark_request) + let benchmark_set_one = BenchmarkSet(0u32); + let collector_name_one = "collector-1"; + db.add_collector_config(collector_name_one, target, benchmark_set_one.0, true) .await .unwrap(); - complete_request(&*db, tag, collector_name, benchmark_set.0, target).await; - // record a couple of errors against the tag - let artifact_id = db.artifact_id(&ArtifactId::Tag(tag.to_string())).await; - - db.record_error(artifact_id, "example-1", "This is an error") - .await; - db.record_error(artifact_id, "example-2", "This is another error") - .await; - - let benchmark_request_two = BenchmarkRequest::create_release(tag_two, time); - db.insert_benchmark_request(&benchmark_request_two) + let benchmark_set_two = BenchmarkSet(1u32); + let collector_name_two = "collector-2"; + db.add_collector_config(collector_name_two, target, benchmark_set_two.0, true) .await .unwrap(); - db.enqueue_benchmark_job( - benchmark_request_two.tag().unwrap(), - target, - CodegenBackend::Llvm, - Profile::Opt, - benchmark_set.0, - ) - .await - .unwrap(); - db.enqueue_benchmark_job( - benchmark_request_two.tag().unwrap(), - target, - CodegenBackend::Llvm, - Profile::Debug, - benchmark_set.0, - ) - .await - .unwrap(); + let collector_configs = db.get_collector_configs().await; + assert!(collector_configs.is_ok()); + let collector_configs = collector_configs.unwrap(); - db.update_benchmark_request_status( - benchmark_request_two.tag().unwrap(), - BenchmarkRequestStatus::InProgress, - ) - .await - .unwrap(); + assert_eq!(collector_configs[0].name(), collector_name_one); + assert_eq!(collector_configs[0].benchmark_set(), benchmark_set_one); + assert!(collector_configs[0].is_active()); - let status_page_data = db.get_status_page_data().await.unwrap(); + assert_eq!(collector_configs[1].name(), collector_name_two); + assert_eq!(collector_configs[1].benchmark_set(), benchmark_set_two); + assert!(collector_configs[1].is_active()); - assert!(status_page_data.completed_requests.len() == 1); - assert_eq!(status_page_data.completed_requests[0].0.tag().unwrap(), tag); - assert!(matches!( - status_page_data.completed_requests[0].0.status(), - BenchmarkRequestStatus::Completed { .. } - )); - // can't really test duration - // ensure errors are correct - assert_eq!( - status_page_data.completed_requests[0].1[0], - "This is an error".to_string() - ); - assert_eq!( - status_page_data.completed_requests[0].1[1], - "This is another error".to_string() - ); + Ok(ctx) + }) + .await; + } - assert!(status_page_data.in_progress.len() == 1); - // we should have 2 jobs - assert!(status_page_data.in_progress[0].request.1.len() == 2); - // the request should be in progress - assert!(matches!( - status_page_data.in_progress[0].request.0.status(), - BenchmarkRequestStatus::InProgress - )); + #[tokio::test] + async fn get_last_completed_requests() { + run_postgres_test(|ctx| async { + let mut requests = vec![]; + let db = ctx.db(); + + let collector = ctx.add_collector(Default::default()).await; + + // Create several completed requests + for id in 1..=3 { + // Make some space between completions + tokio::time::sleep(Duration::from_millis(100)).await; + + requests.push( + RequestBuilder::master( + db, + &format!("sha{}", id), + &format!("sha{}", id - 1), + id, + ) + .await + .add_job(db, job()) + .await + .complete(db, &collector) + .await, + ); + } - // Test the first job - assert!(matches!( - status_page_data.in_progress[0].request.1[0].target(), - Target::X86_64UnknownLinuxGnu - )); - assert!(matches!( - status_page_data.in_progress[0].request.1[0].status(), - BenchmarkJobStatus::Queued - )); - assert!(matches!( - status_page_data.in_progress[0].request.1[0].backend(), - CodegenBackend::Llvm - )); - assert!(matches!( - status_page_data.in_progress[0].request.1[0].profile(), - Profile::Opt - )); - assert_eq!( - status_page_data.in_progress[0].request.1[0].benchmark_set(), - benchmark_set - ); + // Create an additional non-completed request + ctx.insert_master_request("foo", "bar", 1000).await; + + // Request 1 will have artifact with errors + let aid1 = ctx.upsert_master_artifact("sha1").await; + db.record_error(aid1, "crate1", "error1").await; + db.record_error(aid1, "crate2", "error2").await; + + // Request 2 will have artifact without errors + let _aid2 = ctx.upsert_master_artifact("sha2").await; + + // Request 3 will have no artifact (shouldn't happen in practice, but...) + + let reqs = db.get_last_n_completed_benchmark_requests(5).await.unwrap(); + assert_eq!(reqs.len(), 3); + + let expected = [ + ("sha3", HashMap::new()), + ("sha2", HashMap::new()), + ( + "sha1", + HashMap::from([ + ("crate1".to_string(), "error1".to_string()), + ("crate2".to_string(), "error2".to_string()), + ]), + ), + ]; + for ((sha, errors), req) in expected.into_iter().zip(reqs) { + assert_eq!( + req.request.tag().unwrap(), + sha, + "Request {req:?} does not have expected sha {sha}" + ); + assert_eq!( + req.errors, errors, + "Request {req:?} does not have expected errors {errors:?}" + ); + } - // test the second job - assert!(matches!( - status_page_data.in_progress[0].request.1[1].target(), - Target::X86_64UnknownLinuxGnu - )); - assert!(matches!( - status_page_data.in_progress[0].request.1[1].status(), - BenchmarkJobStatus::Queued - )); - assert!(matches!( - status_page_data.in_progress[0].request.1[1].backend(), - CodegenBackend::Llvm - )); - assert!(matches!( - status_page_data.in_progress[0].request.1[1].profile(), - Profile::Debug - )); - assert_eq!( - status_page_data.in_progress[0].request.1[1].benchmark_set(), - benchmark_set - ); + let reqs = db.get_last_n_completed_benchmark_requests(1).await.unwrap(); + assert_eq!(reqs.len(), 1); + assert_eq!(reqs[0].request.tag().unwrap(), "sha3"); Ok(ctx) }) @@ -1155,34 +1138,83 @@ mod tests { } #[tokio::test] - async fn get_collector_configs() { + async fn get_in_progress_jobs() { run_postgres_test(|ctx| async { - let db = ctx.db_client().connection().await; - let target = Target::X86_64UnknownLinuxGnu; + let db = ctx.db(); - let benchmark_set_one = BenchmarkSet(0u32); - let collector_name_one = "collector-1"; - db.add_collector_config(collector_name_one, target, benchmark_set_one.0, true) + let collector = ctx.add_collector(Default::default()).await; + + // Artifacts ready request, should be ignored + RequestBuilder::master(db, "foo", "bar", 1000).await; + + // Create a completed parent with jobs + let completed = RequestBuilder::master(db, "sha4-parent", "sha0", 1001) .await - .unwrap(); + .add_jobs( + db, + &[job().profile(Profile::Doc), job().profile(Profile::Opt)], + ) + .await + .complete(db, &collector) + .await; - let benchmark_set_two = BenchmarkSet(1u32); - let collector_name_two = "collector-2"; - db.add_collector_config(collector_name_two, target, benchmark_set_two.0, true) + // In progress request without a parent + let req1 = RequestBuilder::master(db, "sha1", "sha0", 1) .await - .unwrap(); + .set_in_progress(db) + .await; - let collector_configs = db.get_collector_configs().await; - assert!(collector_configs.is_ok()); - let collector_configs = collector_configs.unwrap(); + // In progress request with a parent that has no jobs + let req2 = RequestBuilder::master(db, "sha2", "sha1", 2) + .await + .add_jobs( + db, + &[job().profile(Profile::Check), job().profile(Profile::Debug)], + ) + .await + .set_in_progress(db) + .await; - assert_eq!(collector_configs[0].name(), collector_name_one); - assert_eq!(collector_configs[0].benchmark_set(), benchmark_set_one); - assert!(collector_configs[0].is_active()); + // In progress request with a parent that has jobs + let req3 = RequestBuilder::master(db, "sha3", "sha2", 3) + .await + .add_jobs( + db, + &[job().profile(Profile::Doc), job().profile(Profile::Opt)], + ) + .await + .set_in_progress(db) + .await; - assert_eq!(collector_configs[1].name(), collector_name_two); - assert_eq!(collector_configs[1].benchmark_set(), benchmark_set_two); - assert!(collector_configs[1].is_active()); + // In progress request with a parent that has jobs, but is completed + let req4 = RequestBuilder::master(db, "sha4", completed.tag(), 4) + .await + .add_jobs( + db, + &[job().profile(Profile::Doc), job().profile(Profile::Check)], + ) + .await + .set_in_progress(db) + .await; + + let mut reqs = db + .get_jobs_of_in_progress_benchmark_requests() + .await + .unwrap(); + + // Check that all jobs are unique + let mut job_ids = HashSet::new(); + for job in reqs.values().flatten() { + assert!(job_ids.insert(job.id)); + } + + // Check that all jobs were returned + assert!(!reqs.contains_key(req1.tag())); + req2.assert_has_exact_jobs(&reqs.remove(req2.tag()).unwrap()); + req3.assert_has_exact_jobs(&reqs.remove(req3.tag()).unwrap()); + req4.assert_has_exact_jobs(&reqs.remove(req4.tag()).unwrap()); + completed.assert_has_exact_jobs(&reqs.remove(completed.tag()).unwrap()); + assert!(reqs.is_empty()); Ok(ctx) }) diff --git a/database/src/pool/postgres.rs b/database/src/pool/postgres.rs index 1bb6adbf1..c5b15092f 100644 --- a/database/src/pool/postgres.rs +++ b/database/src/pool/postgres.rs @@ -3,12 +3,11 @@ use crate::selector::CompileTestCase; use crate::{ ArtifactCollection, ArtifactId, ArtifactIdNumber, Benchmark, BenchmarkJob, BenchmarkJobConclusion, BenchmarkJobStatus, BenchmarkRequest, BenchmarkRequestIndex, - BenchmarkRequestStatus, BenchmarkRequestType, BenchmarkSet, CodegenBackend, CollectionId, - CollectorConfig, Commit, CommitType, CompileBenchmark, Date, InProgressRequestWithJobs, Index, - PartialStatusPageData, Profile, QueuedCommit, Scenario, Target, - BENCHMARK_JOB_STATUS_FAILURE_STR, BENCHMARK_JOB_STATUS_IN_PROGRESS_STR, - BENCHMARK_JOB_STATUS_QUEUED_STR, BENCHMARK_JOB_STATUS_SUCCESS_STR, - BENCHMARK_REQUEST_MASTER_STR, BENCHMARK_REQUEST_RELEASE_STR, + BenchmarkRequestStatus, BenchmarkRequestType, BenchmarkRequestWithErrors, BenchmarkSet, + CodegenBackend, CollectionId, CollectorConfig, Commit, CommitType, CompileBenchmark, Date, + Index, Profile, QueuedCommit, Scenario, Target, BENCHMARK_JOB_STATUS_FAILURE_STR, + BENCHMARK_JOB_STATUS_IN_PROGRESS_STR, BENCHMARK_JOB_STATUS_QUEUED_STR, + BENCHMARK_JOB_STATUS_SUCCESS_STR, BENCHMARK_REQUEST_MASTER_STR, BENCHMARK_REQUEST_RELEASE_STR, BENCHMARK_REQUEST_STATUS_ARTIFACTS_READY_STR, BENCHMARK_REQUEST_STATUS_COMPLETED_STR, BENCHMARK_REQUEST_STATUS_IN_PROGRESS_STR, BENCHMARK_REQUEST_STATUS_WAITING_FOR_ARTIFACTS_STR, BENCHMARK_REQUEST_TRY_STR, @@ -471,6 +470,8 @@ pub struct CachedStatements { get_artifact_size: Statement, load_benchmark_request_index: Statement, get_compile_test_cases_with_measurements: Statement, + get_last_n_completed_requests_with_errors: Statement, + get_jobs_of_in_progress_benchmark_requests: Statement, } pub struct PostgresTransaction<'a> { @@ -667,12 +668,70 @@ impl PostgresConnection { WHERE aid = $1 ) ").await.unwrap(), + get_last_n_completed_requests_with_errors: conn.prepare(&format!(" + WITH completed AS ( + SELECT {BENCHMARK_REQUEST_COLUMNS} + FROM benchmark_request + WHERE status = $1 + -- Select last N completed requests + ORDER BY completed_at DESC + LIMIT $2 + ), artifacts AS ( + SELECT artifact.id, name + FROM artifact + -- Use right join to only return artifacts for selected requests + RIGHT JOIN completed ON artifact.name = completed.tag + ), errors AS ( + SELECT + artifacts.name AS tag, + error.benchmark, + error.error + FROM error + -- Use right join to only return errors for selected artifacts + RIGHT JOIN artifacts ON error.aid = artifacts.id + ) + -- Select request duplicated for each pair of (benchmark, error) + SELECT + completed.*, + errors.benchmark, + errors.error + FROM completed + LEFT JOIN errors ON errors.tag = completed.tag + -- Re-sort the requests, because the original order may be lost + ORDER BY completed.completed_at DESC; + ")).await.unwrap(), + get_jobs_of_in_progress_benchmark_requests: conn.prepare(&format!(" + -- Get in progress requests + WITH in_progress AS ( + SELECT tag, parent_sha + FROM benchmark_request + WHERE status = '{BENCHMARK_REQUEST_STATUS_IN_PROGRESS_STR}' AND + tag IS NOT NULL + ), + -- Get their parents + parents AS ( + SELECT parent_sha AS tag + FROM in_progress + WHERE parent_sha is not NULL + ), + -- Concatenate them together (without duplicates) + requests AS ( + SELECT tag FROM in_progress + UNION + SELECT tag FROM parents + ) + SELECT job_queue.* + FROM requests + -- Only get requests that have some jobs + RIGHT JOIN job_queue on job_queue.request_tag = requests.tag + ")).await.unwrap(), }), conn, } } } +// `tag` should be kept as the first column const BENCHMARK_REQUEST_COLUMNS: &str = "tag, parent_sha, pr, commit_type, status, created_at, completed_at, backends, profiles, commit_date, duration_ms"; @@ -1506,7 +1565,7 @@ where &[ &benchmark_request.tag(), &benchmark_request.parent_sha(), - &benchmark_request.pr().map(|it| *it as i32), + &benchmark_request.pr().map(|it| it as i32), &benchmark_request.commit_type, &benchmark_request.status.as_str(), &benchmark_request.created_at, @@ -1637,9 +1696,10 @@ where backend: CodegenBackend, profile: Profile, benchmark_set: u32, - ) -> anyhow::Result<()> { - self.conn() - .execute( + ) -> anyhow::Result { + let row = self + .conn() + .query_one( r#" INSERT INTO job_queue( request_tag, @@ -1651,6 +1711,7 @@ where ) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT DO NOTHING + RETURNING job_queue.id "#, &[ &request_tag, @@ -1663,7 +1724,7 @@ where ) .await .context("failed to insert benchmark_job")?; - Ok(()) + Ok(row.get::<_, i32>(0) as u32) } async fn get_compile_test_cases_with_measurements( @@ -1902,7 +1963,7 @@ where } } - async fn mark_benchmark_request_as_completed(&self, tag: &str) -> anyhow::Result { + async fn maybe_mark_benchmark_request_as_completed(&self, tag: &str) -> anyhow::Result { // Find if the benchmark is completed and update it's status to completed // in one SQL block let row = self @@ -1984,221 +2045,109 @@ where Ok(()) } - async fn get_status_page_data(&self) -> anyhow::Result { - let max_completed_requests = 30; - - let in_progress_query = format!( - " - WITH in_progress_requests AS ( - SELECT - tag, - parent_sha, - pr, - commit_type, - status, - created_at, - completed_at, - backends, - profiles, - commit_date, - duration_ms - FROM - benchmark_request - WHERE - status = '{BENCHMARK_REQUEST_STATUS_IN_PROGRESS_STR}' - ORDER BY - completed_at - ), in_progress_jobs AS ( - SELECT - request_tag AS tag, - ARRAY_AGG( - ROW( - job_queue.id, - job_queue.request_tag, - job_queue.target, - job_queue.backend, - job_queue.profile, - job_queue.benchmark_set, - job_queue.status, - job_queue.created_at, - job_queue.started_at, - job_queue.completed_at, - job_queue.retry, - job_queue.collector_name - )::TEXT - ) AS jobs - FROM - job_queue - LEFT JOIN in_progress_requests ON job_queue.request_tag = in_progress_requests.tag - GROUP BY - job_queue.request_tag - ), parent AS ( - SELECT - benchmark_request.tag AS parent_tag, - benchmark_request.parent_sha AS parent_sha, - benchmark_request.pr AS parent_pr, - benchmark_request.commit_type AS parent_commit_type, - benchmark_request.status AS parent_status, - benchmark_request.created_at AS parent_created_at, - benchmark_request.completed_at AS parent_completed_at, - benchmark_request.backends AS parent_backends, - benchmark_request.profiles AS parent_profiles, - benchmark_request.commit_date AS parent_commit_date, - benchmark_request.duration_ms AS parent_duration_ms, - EXISTS ( - SELECT - 1 - FROM - job_queue - WHERE - job_queue.request_tag = benchmark_request.tag - AND job_queue.status IN ( - '{BENCHMARK_JOB_STATUS_QUEUED_STR}', - '{BENCHMARK_JOB_STATUS_IN_PROGRESS_STR}' - ) - ) AS parent_active - FROM - benchmark_request - LEFT JOIN - in_progress_requests ON benchmark_request.tag = in_progress_requests.parent_sha - ), parent_jobs AS ( - SELECT - request_tag AS parent_tag, - ARRAY_AGG( - ROW( - job_queue.id, - job_queue.request_tag, - job_queue.target, - job_queue.backend, - job_queue.profile, - job_queue.benchmark_set, - job_queue.status, - job_queue.created_at, - job_queue.started_at, - job_queue.completed_at, - job_queue.retry, - job_queue.collector_name - )::TEXT - ) AS parent_jobs - FROM - job_queue - LEFT JOIN parent ON job_queue.request_tag = parent.parent_tag - GROUP BY - job_queue.request_tag + async fn get_last_n_completed_benchmark_requests( + &self, + count: u64, + ) -> anyhow::Result> { + let rows = self + .conn() + .query( + &self.statements().get_last_n_completed_requests_with_errors, + &[&BENCHMARK_REQUEST_STATUS_COMPLETED_STR, &(count as i64)], ) - SELECT - in_progress_requests.*, - in_progress_jobs.jobs, - parent.*, - parent_jobs.parent_jobs - FROM - in_progress_requests - LEFT JOIN - in_progress_jobs ON in_progress_requests.tag = in_progress_jobs.tag - LEFT JOIN - parent_jobs ON in_progress_requests.parent_sha = parent_jobs.parent_tag - LEFT JOIN - parent ON in_progress_requests.parent_sha = parent.parent_tag;" - ); + .await?; - // Gets requests along with how long the request took (latest job finish - // - earliest job start) and associated errors with the request if they - // exist - let completed_requests_query = format!( - " - WITH completed AS ( - SELECT - {BENCHMARK_REQUEST_COLUMNS} - FROM - benchmark_request - WHERE - status = '{BENCHMARK_REQUEST_STATUS_COMPLETED_STR}' - ORDER BY - completed_at - DESC LIMIT {max_completed_requests} - ), artifacts AS ( - SELECT - artifact.id, - name - FROM - artifact - LEFT JOIN completed ON artifact.name = completed.tag - ), errors AS ( - SELECT - artifacts.name AS tag, - ARRAY_AGG(error) AS errors - FROM - error - LEFT JOIN - artifacts ON error.aid = artifacts.id - GROUP BY - tag - ) - SELECT - completed.*, - errors.errors AS errors - FROM - completed - LEFT JOIN errors ON errors.tag = completed.tag; - " - ); + // Iterate through the requests and aggregate their errors + // Make sure to keep their original order + let mut requests = vec![]; + // tag -> errors + let mut errors: HashMap> = Default::default(); - let in_progress: Vec = self - .conn() - .query(&in_progress_query, &[]) - .await? - .iter() - .map(|it| { - let benchmark_request = row_to_benchmark_request(it, None); - let jobs: Vec = it - .get::<_, Vec>("jobs") - .iter() - .map(|it| benchmark_job_str_to_type(it).unwrap()) - .collect(); - - // This is ever-so-slightly grim however it allows us to not - // have to parse the text representation of the jobs. Which - // saves a reasonable amount of time to justify doing this. - let parent_active = it.get::<_, Option>("parent_active"); - - InProgressRequestWithJobs { - request: (benchmark_request, jobs), - parent: if parent_active.unwrap_or(false) { - // The rows values will only be non-null if the `parent_active` - // has been set - let parent_benchmark_request = row_to_benchmark_request(it, Some(12)); - // Only parse the jobs if we need to include the parent - let parent_jobs: Vec = it - .get::<_, Vec>("parent_jobs") - .iter() - .map(|it| benchmark_job_str_to_type(it).unwrap()) - .collect(); - Some((parent_benchmark_request, parent_jobs)) - } else { - None - }, + for row in rows { + let tag = row.get::<_, &str>(0); + let error_benchmark = row.get::<_, Option>(11); + let error_content = row.get::<_, Option>(12); + + // We already saw this request, just add errors + if let Some(errors) = errors.get_mut(tag) { + if let Some(benchmark) = error_benchmark { + errors.insert(benchmark, error_content.unwrap_or_default()); } + } else { + // We see this request for the first time + let request = row_to_benchmark_request(&row, None); + let request_errors = if let Some(benchmark) = error_benchmark { + HashMap::from([(benchmark, error_content.unwrap_or_default())]) + } else { + HashMap::new() + }; + errors.insert(tag.to_string(), request_errors); + requests.push(request); + } + } + + Ok(requests + .into_iter() + .map(|request| { + let errors = errors.remove(request.tag().unwrap()).unwrap_or_default(); + BenchmarkRequestWithErrors { request, errors } }) - .collect(); + .collect()) + } - let completed_requests: Vec<(BenchmarkRequest, Vec)> = self + async fn get_jobs_of_in_progress_benchmark_requests( + &self, + ) -> anyhow::Result>> { + let rows = self .conn() - .query(&completed_requests_query, &[]) - .await? - .iter() - .map(|it| { - ( - row_to_benchmark_request(it, None), - // The errors, if there are none this will be an empty vector - it.get::<_, Option>>(11).unwrap_or_default(), - ) - }) - .collect(); + .query( + &self.statements().get_jobs_of_in_progress_benchmark_requests, + &[], + ) + .await?; - Ok(PartialStatusPageData { - completed_requests, - in_progress, - }) + let mut request_to_jobs: HashMap> = HashMap::new(); + for row in rows { + let started_at = row.get::<_, Option>>(7); + let status = row.get::<_, &str>(9); + let collector_name = row.get::<_, Option>(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>(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>(6), + status, + deque_counter: row.get::<_, i32>(10) as u32, + }; + request_to_jobs + .entry(job.request_tag.clone()) + .or_default() + .push(job); + } + Ok(request_to_jobs) } async fn get_collector_configs(&self) -> anyhow::Result> { @@ -2271,7 +2220,9 @@ fn row_to_benchmark_request(row: &Row, row_offset: Option) -> BenchmarkRe let status = BenchmarkRequestStatus::from_str_and_completion_date(status, completed_at, duration_ms) - .expect("Invalid BenchmarkRequestStatus data in the database"); + .unwrap_or_else(|e| { + panic!("Invalid BenchmarkRequestStatus data in the database for tag {tag:?}: {e:?}") + }); match commit_type { BENCHMARK_REQUEST_TRY_STR => BenchmarkRequest { @@ -2312,96 +2263,6 @@ fn row_to_benchmark_request(row: &Row, row_offset: Option) -> BenchmarkRe } } -fn parse_timestamp(cell: &str) -> anyhow::Result>> { - if cell.is_empty() { - Ok(None) - } else { - // Massage postgres date string into something we can parse in Rust - // to a date - let raw_date = cell.trim_matches('"').replace(' ', "T") + ":00"; - Ok(Some( - DateTime::parse_from_rfc3339(&raw_date)?.with_timezone(&Utc), - )) - } -} - -fn benchmark_job_str_to_type(src: &str) -> anyhow::Result { - let line = src.trim_start_matches('(').trim_end_matches(')'); - - let mut col = line.split(','); - - let id: u32 = col.next().ok_or_else(|| anyhow::anyhow!("id"))?.parse()?; - let request_tag = col - .next() - .ok_or_else(|| anyhow::anyhow!("request_tag"))? - .to_owned(); - let target = col - .next() - .ok_or_else(|| anyhow::anyhow!("target"))? - .parse::() - .map_err(|e| anyhow::anyhow!(e))?; - let backend = col - .next() - .ok_or_else(|| anyhow::anyhow!("backend"))? - .parse::() - .map_err(|e| anyhow::anyhow!(e))?; - let profile = col - .next() - .ok_or_else(|| anyhow::anyhow!("profile"))? - .parse::() - .map_err(|e| anyhow::anyhow!(e))?; - let benchmark_set = BenchmarkSet( - col.next() - .ok_or_else(|| anyhow::anyhow!("benchmark_set"))? - .parse()?, - ); - - let status_str = col.next().ok_or_else(|| anyhow::anyhow!("status"))?; - let created_at = parse_timestamp(col.next().ok_or_else(|| anyhow::anyhow!("created_at"))?)? - .ok_or_else(|| anyhow::anyhow!("created_at missing"))?; - - let started_at = parse_timestamp(col.next().unwrap_or(""))?; - let completed_at = parse_timestamp(col.next().unwrap_or(""))?; - let retry: u32 = col - .next() - .ok_or_else(|| anyhow::anyhow!("retry"))? - .parse()?; - let collector_name_raw = col.next().unwrap_or("").to_owned(); - - let status = match status_str { - BENCHMARK_JOB_STATUS_QUEUED_STR => BenchmarkJobStatus::Queued, - - BENCHMARK_JOB_STATUS_IN_PROGRESS_STR => BenchmarkJobStatus::InProgress { - started_at: started_at.ok_or_else(|| anyhow::anyhow!("started_at missing"))?, - collector_name: collector_name_raw, - }, - - BENCHMARK_JOB_STATUS_SUCCESS_STR | BENCHMARK_JOB_STATUS_FAILURE_STR => { - BenchmarkJobStatus::Completed { - started_at: started_at.ok_or_else(|| anyhow::anyhow!("started_at missing"))?, - completed_at: completed_at - .ok_or_else(|| anyhow::anyhow!("completed_at missing"))?, - collector_name: collector_name_raw, - success: status_str == BENCHMARK_JOB_STATUS_SUCCESS_STR, - } - } - - _ => anyhow::bail!("unknown status `{status_str}`"), - }; - - Ok(BenchmarkJob { - id, - target, - backend, - profile, - request_tag, - benchmark_set, - created_at, - status, - deque_counter: retry, - }) -} - fn parse_artifact_id(ty: &str, sha: &str, date: Option>) -> ArtifactId { match ty { "master" => ArtifactId::Commit(Commit { diff --git a/database/src/pool/sqlite.rs b/database/src/pool/sqlite.rs index a0e2a19b3..1d69e133d 100644 --- a/database/src/pool/sqlite.rs +++ b/database/src/pool/sqlite.rs @@ -2,9 +2,9 @@ use crate::pool::{Connection, ConnectionManager, ManagedConnection, Transaction} use crate::selector::CompileTestCase; use crate::{ ArtifactCollection, ArtifactId, Benchmark, BenchmarkJob, BenchmarkJobConclusion, - BenchmarkRequest, BenchmarkRequestIndex, BenchmarkRequestStatus, BenchmarkSet, CodegenBackend, - CollectionId, CollectorConfig, Commit, CommitType, CompileBenchmark, Date, - PartialStatusPageData, Profile, Target, + BenchmarkRequest, BenchmarkRequestIndex, BenchmarkRequestStatus, BenchmarkRequestWithErrors, + BenchmarkSet, CodegenBackend, CollectionId, CollectorConfig, Commit, CommitType, + CompileBenchmark, Date, Profile, Target, }; use crate::{ArtifactIdNumber, Index, QueuedCommit}; use chrono::{DateTime, TimeZone, Utc}; @@ -1301,7 +1301,7 @@ impl Connection for SqliteConnection { _backend: CodegenBackend, _profile: Profile, _benchmark_set: u32, - ) -> anyhow::Result<()> { + ) -> anyhow::Result { no_queue_implementation_abort!() } @@ -1359,7 +1359,7 @@ impl Connection for SqliteConnection { no_queue_implementation_abort!() } - async fn mark_benchmark_request_as_completed(&self, _tag: &str) -> anyhow::Result { + async fn maybe_mark_benchmark_request_as_completed(&self, _tag: &str) -> anyhow::Result { no_queue_implementation_abort!() } @@ -1371,7 +1371,9 @@ impl Connection for SqliteConnection { no_queue_implementation_abort!() } - async fn get_status_page_data(&self) -> anyhow::Result { + async fn get_jobs_of_in_progress_benchmark_requests( + &self, + ) -> anyhow::Result>> { no_queue_implementation_abort!() } @@ -1382,6 +1384,13 @@ impl Connection for SqliteConnection { async fn update_collector_heartbeat(&self, _collector_name: &str) -> anyhow::Result<()> { no_queue_implementation_abort!() } + + async fn get_last_n_completed_benchmark_requests( + &self, + _count: u64, + ) -> anyhow::Result> { + no_queue_implementation_abort!() + } } fn parse_artifact_id(ty: &str, sha: &str, date: Option) -> ArtifactId { diff --git a/database/src/tests/builder.rs b/database/src/tests/builder.rs new file mode 100644 index 000000000..48679e657 --- /dev/null +++ b/database/src/tests/builder.rs @@ -0,0 +1,151 @@ +use crate::{ + BenchmarkJob, BenchmarkJobConclusion, BenchmarkRequest, BenchmarkRequestStatus, BenchmarkSet, + CodegenBackend, CollectorConfig, Connection, Profile, Target, +}; +use chrono::Utc; +use hashbrown::{HashMap, HashSet}; + +pub struct RequestBuilder { + request: BenchmarkRequest, + jobs: Vec<(JobBuilder, u32)>, +} + +impl RequestBuilder { + pub async fn master(db: &dyn Connection, tag: &str, parent: &str, pr: u32) -> Self { + let request = BenchmarkRequest::create_master(tag, parent, pr, Utc::now()); + db.insert_benchmark_request(&request).await.unwrap(); + Self { + request, + jobs: vec![], + } + } + + pub fn tag(&self) -> &str { + self.request.tag().unwrap() + } + + pub fn assert_has_exact_jobs(&self, jobs: &[BenchmarkJob]) { + assert_eq!(jobs.len(), self.jobs.len()); + let mut expected: HashSet = self.jobs.iter().map(|(_, id)| *id).collect(); + for job in jobs { + assert!(expected.remove(&job.id)); + } + assert!(expected.is_empty()); + } + + pub async fn add_job(self, db: &dyn Connection, job: JobBuilder) -> Self { + self.add_jobs(db, &[job]).await + } + + pub async fn add_jobs(mut self, db: &dyn Connection, jobs: &[JobBuilder]) -> Self { + for job in jobs { + let id = db + .enqueue_benchmark_job( + self.tag(), + job.target, + job.backend, + job.profile, + job.benchmark_set, + ) + .await + .unwrap(); + self.jobs.push((job.clone(), id)); + } + self + } + + pub async fn set_in_progress(self, db: &dyn Connection) -> Self { + db.update_benchmark_request_status(self.tag(), BenchmarkRequestStatus::InProgress) + .await + .unwrap(); + self + } + + /// Continually completes **pending jobs in the DB** until all jobs of this request are + /// completed, and then completes this benchmark request. + pub async fn complete(self, db: &dyn Connection, collector: &CollectorConfig) -> Self { + assert!(!self.jobs.is_empty()); + let tag = self.tag().to_string(); + + let mut to_complete: HashMap = + self.jobs.iter().map(|(job, id)| (*id, job)).collect(); + while !to_complete.is_empty() { + // We can't specify which job we dequeue, so we have to iterate them one by one and + // complete them, until we complete all the jobs that we expect + let (target, set) = to_complete + .values() + .map(|j| (j.target, j.benchmark_set)) + .next() + .unwrap(); + let (job, _) = db + .dequeue_benchmark_job(collector.name(), target, BenchmarkSet(set)) + .await + .unwrap() + .unwrap(); + let conclusion = if let Some(expected_job) = to_complete.remove(&job.id) { + expected_job.conclusion.clone() + } else { + BenchmarkJobConclusion::Success + }; + db.mark_benchmark_job_as_completed(job.id, conclusion) + .await + .unwrap(); + } + // At this point all jobs of the request should be properly completed, so we can also + // complete the request itself + assert!(db + .maybe_mark_benchmark_request_as_completed(&tag) + .await + .unwrap()); + drop(to_complete); + self + } +} + +#[derive(Clone)] +pub struct JobBuilder { + target: Target, + backend: CodegenBackend, + profile: Profile, + benchmark_set: u32, + conclusion: BenchmarkJobConclusion, +} + +impl JobBuilder { + pub fn profile(mut self, profile: Profile) -> Self { + self.profile = profile; + self + } +} + +impl Default for JobBuilder { + fn default() -> Self { + Self { + target: Target::X86_64UnknownLinuxGnu, + backend: CodegenBackend::Llvm, + profile: Profile::Check, + benchmark_set: 0, + conclusion: BenchmarkJobConclusion::Success, + } + } +} + +pub fn job() -> JobBuilder { + JobBuilder::default() +} + +pub struct CollectorBuilder { + pub name: String, + pub target: Target, + pub benchmark_set: BenchmarkSet, +} + +impl Default for CollectorBuilder { + fn default() -> Self { + Self { + name: "test-collector".to_string(), + target: Target::X86_64UnknownLinuxGnu, + benchmark_set: BenchmarkSet(0), + } + } +} diff --git a/database/src/tests/mod.rs b/database/src/tests/mod.rs index 6b411a9d1..6beacd32b 100644 --- a/database/src/tests/mod.rs +++ b/database/src/tests/mod.rs @@ -1,11 +1,18 @@ #![allow(dead_code)] +pub mod builder; + +use chrono::Utc; use std::future::Future; use tokio_postgres::config::Host; use tokio_postgres::Config; use crate::pool::postgres::make_client; -use crate::Pool; +use crate::tests::builder::CollectorBuilder; +use crate::{ + ArtifactId, ArtifactIdNumber, BenchmarkRequest, CollectorConfig, Commit, CommitType, + Connection, Date, Pool, +}; enum TestDb { Postgres { @@ -20,10 +27,12 @@ enum TestDb { /// a database. pub struct TestContext { test_db: TestDb, - // Pre-cached client to avoid creating unnecessary connections in tests - client: Pool, + pool: Pool, + // Pre-cached DB connection + connection: Box, } +/// Basic lifecycle functions impl TestContext { async fn new_postgres(db_url: &str) -> Self { let config: Config = db_url.parse().expect("Cannot parse connection string"); @@ -65,32 +74,37 @@ impl TestContext { db_name ); let pool = Pool::open(test_db_url.as_str()); + let connection = pool.connection().await; Self { test_db: TestDb::Postgres { original_db_url: db_url.to_string(), db_name, }, - client: pool, + pool, + connection, } } async fn new_sqlite() -> Self { let pool = Pool::open(":memory:"); + let connection = pool.connection().await; Self { test_db: TestDb::SQLite, - client: pool, + pool, + connection, } } - pub fn db_client(&self) -> &Pool { - &self.client + pub fn db(&self) -> &dyn Connection { + self.connection.as_ref() } async fn finish(self) { // Cleanup the test database // First, we need to stop using the database - drop(self.client); + drop(self.connection); + drop(self.pool); match self.test_db { TestDb::Postgres { @@ -111,6 +125,51 @@ impl TestContext { } } +/// Test helpers +impl TestContext { + /// Create a new master benchmark request and add it to the DB. + pub async fn insert_master_request( + &self, + sha: &str, + parent: &str, + pr: u32, + ) -> BenchmarkRequest { + let req = BenchmarkRequest::create_master(sha, parent, pr, Utc::now()); + self.db().insert_benchmark_request(&req).await.unwrap(); + req + } + + pub async fn complete_request(&self, tag: &str) { + // Note: this assumes that there are not non-completed jobs in the DB for the request + self.db() + .maybe_mark_benchmark_request_as_completed(tag) + .await + .unwrap(); + } + + pub async fn upsert_master_artifact(&self, sha: &str) -> ArtifactIdNumber { + self.db() + .artifact_id(&ArtifactId::Commit(Commit { + sha: sha.to_string(), + date: Date(Utc::now()), + r#type: CommitType::Master, + })) + .await + } + + pub async fn add_collector(&self, collector: CollectorBuilder) -> CollectorConfig { + self.db() + .add_collector_config( + &collector.name, + collector.target, + collector.benchmark_set.get_id(), + true, + ) + .await + .unwrap() + } +} + /// Runs a test against an actual postgres database. pub async fn run_postgres_test(f: F) where diff --git a/site/frontend/src/pages/status_new/collector.vue b/site/frontend/src/pages/status_new/collector.vue index 3123ecf00..68a8f5d00 100644 --- a/site/frontend/src/pages/status_new/collector.vue +++ b/site/frontend/src/pages/status_new/collector.vue @@ -1,18 +1,13 @@ diff --git a/site/frontend/src/pages/status_new/data.ts b/site/frontend/src/pages/status_new/data.ts index 4066b4934..27b9aa81b 100644 --- a/site/frontend/src/pages/status_new/data.ts +++ b/site/frontend/src/pages/status_new/data.ts @@ -1,142 +1,28 @@ -export const BenchmarkRequestCompleteStr = "completed"; -export const BenchmarkRequestInProgressStr = "in_progress"; -export const BenchmarkRequestArtifactsReadyStr = "artifacts_ready"; +export type BenchmarkRequestType = "Release" | "Master" | "Try"; +export type BenchmarkRequestStatus = "Queued" | "InProgress" | "Completed"; -type BenchmarkRequestStatusComplete = { - state: typeof BenchmarkRequestCompleteStr; - completedAt: string; - duration_s: number; -}; - -type BenchmarkRequestStatusInProgress = { - state: typeof BenchmarkRequestInProgressStr; -}; - -type BenchmarkRequestStatusArtifactsReady = { - state: typeof BenchmarkRequestArtifactsReadyStr; -}; - -export type BenchmarkRequestStatus = - | BenchmarkRequestStatusComplete - | BenchmarkRequestStatusInProgress - | BenchmarkRequestStatusArtifactsReady; - -export const TryCommit = "Try"; -export const MasterCommit = "Master"; -export const ReleaseCommit = "Release"; - -type BenchmarkRequestTypeTry = { - type: typeof TryCommit; - tag: string | null; - parent_sha: string | null; - pr: number; -}; - -type BenchmarkRequestTypeMaster = { - type: typeof MasterCommit; +export type BenchmarkRequest = { tag: string; - parent_sha: string; - pr: number; -}; - -type BenchmarkRequestTypeRelease = { - type: typeof ReleaseCommit; - tag: string; -}; - -export type BenchmarkRequestTypeStr = - | typeof ReleaseCommit - | typeof MasterCommit - | typeof TryCommit; - -export type BenchmarkRequestType = - | BenchmarkRequestTypeTry - | BenchmarkRequestTypeMaster - | BenchmarkRequestTypeRelease; - -export type BenchmarkRequestComplete = { - status: BenchmarkRequestStatusComplete; - requestType: BenchmarkRequestType; - commitDate: string | null; - createdAt: string | null; - backends: string[]; - profiles: string; - errors: string[]; -}; - -export type BenchmarkRequestInProgress = { - status: BenchmarkRequestStatusInProgress; + pr: number | null; + status: BenchmarkRequestStatus; requestType: BenchmarkRequestType; - commitDate: string | null; - createdAt: string | null; - backends: string[]; - profiles: string; - errors: string[]; -}; - -export type BenchmarkRequestArtifactsReady = { - status: BenchmarkRequestStatusArtifactsReady; - requestType: BenchmarkRequestType; - commitDate: string | null; - createdAt: string | null; - backends: string[]; - profiles: string; - errors: string[]; -}; - -export type BenchmarkRequest = - | BenchmarkRequestComplete - | BenchmarkRequestInProgress - | BenchmarkRequestArtifactsReady; - -export const BenchmarkJobQueued = "queued"; -export const BenchmarkJobInProgress = "in_progres"; -export const BenchmarkJobFailed = "failed"; -export const BenchmarkJobSuccess = "success"; - -export type BenchmarkJobStatusQueued = { - state: typeof BenchmarkJobQueued; -}; - -export type BenchmarkJobStatusInProgress = { - state: typeof BenchmarkJobInProgress; - startedAt: string; - collectorName: string; -}; - -export type BenchmarkJobStatusFailed = { - state: typeof BenchmarkJobFailed; - startedAt: string; - completedAt: string; - collectorName: string; -}; - -export type BenchmarkJobStatusSuccess = { - state: typeof BenchmarkJobSuccess; - startedAt: string; - completedAt: string; - collectorName: string; + createdAt: string; + completedAt: string | null; + durationS: number | null; + errors: Dict; }; -export type BenchmarkJobStatusStr = - | typeof BenchmarkJobQueued - | typeof BenchmarkJobInProgress - | typeof BenchmarkJobFailed - | typeof BenchmarkJobSuccess; - -export type BenchmarkJobStatus = - | BenchmarkJobStatusSuccess - | BenchmarkJobStatusFailed - | BenchmarkJobStatusInProgress - | BenchmarkJobStatusQueued; +export type BenchmarkJobStatus = "Queued" | "InProgress" | "Success" | "Failed"; export type BenchmarkJob = { + requestTag: string; target: string; backend: string; profile: string; - requestTag: string; benchmarkSet: number; createdAt: string; + startedAt: string | null; + completedAt: string | null; status: BenchmarkJobStatus; dequeCounter: number; }; @@ -148,17 +34,10 @@ export type CollectorConfig = { isActive: boolean; lastHeartbeatAt: string; dateAdded: string; -}; - -export type CollectorInfo = { - config: CollectorConfig; - jobIds: number[]; + jobs: BenchmarkJob[]; }; export type StatusResponse = { - queueRequestTags: string[]; - requestsMap: Dict; - jobMap: Dict; - collectorWorkMap: Dict; - tagToJobs: Dict; + requests: BenchmarkRequest[]; + collectors: CollectorConfig[]; }; diff --git a/site/frontend/src/pages/status_new/page.vue b/site/frontend/src/pages/status_new/page.vue index f245ee152..1487ec1d9 100644 --- a/site/frontend/src/pages/status_new/page.vue +++ b/site/frontend/src/pages/status_new/page.vue @@ -6,50 +6,29 @@ import {STATUS_DATA_NEW_URL} from "../../urls"; import {withLoading} from "../../utils/loading"; import {formatSecondsAsDuration} from "../../utils/formatting"; import { - StatusResponse, - BenchmarkRequestType, BenchmarkRequest, - BenchmarkJob, - CollectorInfo, - ReleaseCommit, - BenchmarkRequestCompleteStr, - BenchmarkRequestInProgressStr, + BenchmarkRequestStatus, + CollectorConfig, + StatusResponse, } from "./data"; import Collector from "./collector.vue"; const loading = ref(true); -const dataNew: Ref<{ - queueLength: number; +const data: Ref<{ timeline: BenchmarkRequestWithWaterLine[]; - requestsMap: Dict; - jobMap: Dict; - collectorWorkMap: Dict; - tagToJobs: Dict; + queueLength: number; + collectors: CollectorConfig[]; } | null> = ref(null); -type BenchmarkRequestWithWaterLine = BenchmarkRequest & {isWaterLine: boolean}; +type BenchmarkRequestWithWaterLine = BenchmarkRequest & { + isLastInProgress: boolean; + hasPendingJobs: boolean; +}; -function requestIsInProgress(req: BenchmarkRequest, tagToJobs: Dict) { - switch (req.status.state) { - case BenchmarkRequestCompleteStr: - if (req.requestType.tag in tagToJobs) { - return true; - } - return false; - case BenchmarkRequestInProgressStr: - return true; - default: - return false; - } -} - -function getRequestRowClassName( - req: BenchmarkRequestWithWaterLine, - tagToJobs: Dict -) { - const inProgress = requestIsInProgress(req, tagToJobs); - if (inProgress && req.isWaterLine) { +function getRequestRowClassName(req: BenchmarkRequestWithWaterLine) { + const inProgress = req.status === "InProgress"; + if (inProgress && req.isLastInProgress) { return "timeline-waterline"; } else if (inProgress) { return "timeline-row-bold"; @@ -57,79 +36,95 @@ function getRequestRowClassName( return ""; } -async function loadStatusNew(loading: Ref) { - dataNew.value = await withLoading(loading, async () => { - let d: StatusResponse = await getJson(STATUS_DATA_NEW_URL); +async function loadStatusData(loading: Ref) { + data.value = await withLoading(loading, async () => { + let resp: StatusResponse = await getJson( + STATUS_DATA_NEW_URL + ); let timeline: BenchmarkRequestWithWaterLine[] = []; - // figure out where to draw the line. - for (let i = 1; i < d.queueRequestTags.length; ++i) { - let req = d.requestsMap[d.queueRequestTags[i - 1]]; - let nextReq = d.requestsMap[d.queueRequestTags[i]]; - let isWaterLine = false; - if ( - requestIsInProgress(req, d.tagToJobs) && - !requestIsInProgress(nextReq, d.tagToJobs) - ) { - isWaterLine = true; + + let queueLength = 0; + + let requests_with_pending_jobs = new Set(); + for (const job of resp.collectors.flatMap((c) => c.jobs)) { + if (job.status === "Queued" || job.status === "InProgress") { + requests_with_pending_jobs.add(job.requestTag); } + } + + // Figure out where to draw the line. + for (let i = 0; i < resp.requests.length; i++) { + let request = resp.requests[i]; + let isLastInProgress = + request.status === "InProgress" && + (i == resp.requests.length - 1 || + resp.requests[i + 1].status !== "InProgress"); timeline.push({ - ...req, - isWaterLine, + ...request, + isLastInProgress, + hasPendingJobs: requests_with_pending_jobs.has(request.tag), }); + + if (request.status !== "Completed") { + queueLength += 1; + } } + return { - queueLength: d.queueRequestTags.length, timeline, - requestsMap: d.requestsMap, - jobMap: d.jobMap, - collectorWorkMap: d.collectorWorkMap, - tagToJobs: d.tagToJobs, + collectors: resp.collectors, + queueLength, }; }); } -function getCreatedAt(request: BenchmarkRequest): string { - if (request.status.state == BenchmarkRequestCompleteStr) { - return request.status.completedAt; +function getDuration(request: BenchmarkRequest): string { + if (request.status === "Completed") { + return formatSecondsAsDuration(request.durationS); } return ""; } -function getDuration(request: BenchmarkRequest): string { - if (request.status.state == BenchmarkRequestCompleteStr) { - return formatSecondsAsDuration(request.status.duration_s); +function formatStatus(status: BenchmarkRequestStatus): string { + if (status === "Completed") { + return "Finished"; + } else if (status === "InProgress") { + return "In progress"; + } else if (status === "Queued") { + return "Queued"; + } else { + return "Unknown"; } - return ""; } -function PullRequestLink({requestType}: {requestType: BenchmarkRequestType}) { - if (requestType.type === ReleaseCommit) { +function PullRequestLink({request}: {request: BenchmarkRequest}) { + if (request.requestType === "Release") { return ""; } return ( - - #{requestType.pr} + + #{request.pr} ); } -loadStatusNew(loading); +loadStatusData(loading);