mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-12 00:06:14 +00:00
fix: reduce DB pool contention by eliminating dual-connection patterns (#7861)
This commit is contained in:
Ruben Fiszel
GitHub
+5
-5
@@ -46,11 +46,11 @@
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
true,
|
||||
true,
|
||||
true,
|
||||
true,
|
||||
true,
|
||||
true
|
||||
]
|
||||
|
||||
+2
-1
@@ -30,7 +30,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -122,7 +122,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -40,7 +40,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
-15
@@ -1,15 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "WITH inserted_concurrency_counter AS (\n INSERT INTO concurrency_counter (concurrency_id, job_uuids) \n VALUES ($1, '{}'::jsonb)\n ON CONFLICT DO NOTHING\n )\n INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Uuid"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514"
|
||||
}
|
||||
+4
-2
@@ -34,7 +34,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -67,7 +68,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -40,7 +40,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -30,7 +30,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -37,7 +37,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+4
-2
@@ -32,7 +32,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -70,7 +71,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "WITH inserted_concurrency_counter AS (\n INSERT INTO concurrency_counter (concurrency_id, job_uuids)\n VALUES ($1, '{}'::jsonb)\n ON CONFLICT DO NOTHING\n )\n INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Uuid"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "5a9cf9cc229f7b7ddbee2485cda4e9f7a91240a00f10d24d3f28d5b722f68768"
|
||||
}
|
||||
+2
-1
@@ -245,7 +245,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -35,7 +35,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -29,7 +29,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -40,7 +40,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -27,7 +27,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -35,7 +35,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -32,7 +32,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -30,7 +30,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -155,7 +155,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "job_id",
|
||||
"type_info": "Uuid"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": []
|
||||
},
|
||||
"nullable": [
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec"
|
||||
}
|
||||
+2
-1
@@ -185,7 +185,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -160,7 +160,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
-14
@@ -1,14 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "DELETE FROM job_result_stream_v2 WHERE job_id = $1",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Uuid"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090"
|
||||
}
|
||||
+2
-1
@@ -105,7 +105,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -31,7 +31,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "DELETE FROM job_perms\nWHERE job_id NOT IN (SELECT id FROM v2_job_queue)\nRETURNING job_id",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "job_id",
|
||||
"type_info": "Uuid"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": []
|
||||
},
|
||||
"nullable": [
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "c825fa5c6e287068aeaad994c0b42b8ad59b9129f032c6b918c27426ab304f2b"
|
||||
}
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -105,7 +105,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -25,7 +25,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -185,7 +185,8 @@
|
||||
"sqs",
|
||||
"gcp",
|
||||
"mqtt",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -31,7 +31,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -24,7 +24,8 @@
|
||||
"mqtt",
|
||||
"gcp",
|
||||
"default_email",
|
||||
"nextcloud"
|
||||
"nextcloud",
|
||||
"google"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, concurrency_key, log_file, metrics",
|
||||
"query": "VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, job_perms, concurrency_key, log_file, metrics",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
@@ -8,5 +8,5 @@
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571"
|
||||
"hash": "f0070b36f7c4fc84dc9c23bb6c73d8ba80993a28b2c2e5df70968acf6d7cebe4"
|
||||
}
|
||||
+46
-2
@@ -1103,7 +1103,6 @@ async fn delete_expired_jobs_batch(
|
||||
job_retention_secs
|
||||
);
|
||||
|
||||
// Delete related records for this batch
|
||||
if let Err(e) = sqlx::query!(
|
||||
"DELETE FROM job_stats WHERE job_id = ANY($1)",
|
||||
&deleted_jobs
|
||||
@@ -1783,6 +1782,19 @@ pub async fn monitor_db(
|
||||
}
|
||||
};
|
||||
|
||||
let cleanup_job_live_rows_f = async {
|
||||
if server_mode && !initial_load {
|
||||
if let Some(db) = conn.as_sql() {
|
||||
if let Err(e) = cleanup_job_perms_orphaned(&db).await {
|
||||
tracing::error!("Error cleaning up orphaned job_perms: {:?}", e);
|
||||
}
|
||||
if let Err(e) = cleanup_job_result_stream_orphaned_jobs(&db).await {
|
||||
tracing::error!("Error cleaning up orphaned job_result_stream_v2: {:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// run every hour (60 minutes / 30 seconds = 120)
|
||||
let cleanup_worker_group_stats_f = async {
|
||||
if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(120) {
|
||||
@@ -1956,6 +1968,7 @@ pub async fn monitor_db(
|
||||
cleanup_debounce_keys_f,
|
||||
cleanup_debounce_keys_completed_f,
|
||||
cleanup_flow_iterator_data_f,
|
||||
cleanup_job_live_rows_f,
|
||||
cleanup_worker_group_stats_f,
|
||||
native_triggers_sync_f,
|
||||
cleanup_notify_events_f,
|
||||
@@ -1963,7 +1976,7 @@ pub async fn monitor_db(
|
||||
}
|
||||
|
||||
async fn vacuuming_tables(db: &Pool<Postgres>) -> error::Result<()> {
|
||||
sqlx::query!("VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, concurrency_key, log_file, metrics")
|
||||
sqlx::query!("VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, job_perms, concurrency_key, log_file, metrics")
|
||||
.execute(db)
|
||||
.await?;
|
||||
Ok(())
|
||||
@@ -3062,6 +3075,37 @@ RETURNING key,job_id
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cleanup_job_perms_orphaned(db: &DB) -> error::Result<()> {
|
||||
let result = sqlx::query_scalar!(
|
||||
"DELETE FROM job_perms
|
||||
WHERE job_id NOT IN (SELECT id FROM v2_job_queue)
|
||||
RETURNING job_id"
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
|
||||
if !result.is_empty() {
|
||||
tracing::info!("Cleaned up {} orphaned job_perms rows", result.len());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cleanup_job_result_stream_orphaned_jobs(db: &DB) -> error::Result<()> {
|
||||
let result = sqlx::query!(
|
||||
"DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id",
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
|
||||
if result.len() > 0 {
|
||||
tracing::info!(
|
||||
"Cleaned up {} orphaned job_result_stream_v2 rows",
|
||||
result.len()
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cleanup_flow_iterator_data_orphaned_jobs(db: &DB) -> error::Result<()> {
|
||||
let result = sqlx::query!(
|
||||
"
|
||||
|
||||
@@ -149,11 +149,17 @@ pub struct RunJob {
|
||||
pub payload: JobPayload,
|
||||
pub args: serde_json::Map<String, serde_json::Value>,
|
||||
pub scheduled_for_o: Option<chrono::DateTime<chrono::Utc>>,
|
||||
pub email: String,
|
||||
}
|
||||
|
||||
impl From<JobPayload> for RunJob {
|
||||
fn from(payload: JobPayload) -> Self {
|
||||
Self { payload, args: Default::default(), scheduled_for_o: None }
|
||||
Self {
|
||||
payload,
|
||||
args: Default::default(),
|
||||
scheduled_for_o: None,
|
||||
email: "test@windmill.dev".to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -171,8 +177,13 @@ impl RunJob {
|
||||
self
|
||||
}
|
||||
|
||||
pub fn email(mut self, email: impl Into<String>) -> Self {
|
||||
self.email = email.into();
|
||||
self
|
||||
}
|
||||
|
||||
pub async fn push(self, db: &Pool<Postgres>) -> Uuid {
|
||||
let RunJob { payload, args, scheduled_for_o } = self;
|
||||
let RunJob { payload, args, scheduled_for_o, email } = self;
|
||||
let mut hm_args = std::collections::HashMap::new();
|
||||
for (k, v) in args {
|
||||
hm_args.insert(k, windmill_common::worker::to_raw_value(&v));
|
||||
@@ -186,7 +197,7 @@ impl RunJob {
|
||||
payload,
|
||||
windmill_queue::PushArgs::from(&hm_args),
|
||||
/* user */ "test-user",
|
||||
/* email */ "test@windmill.dev",
|
||||
/* email */ &email,
|
||||
/* permissioned_as */ "u/test-user".to_string(),
|
||||
/* token_prefix */ None,
|
||||
scheduled_for_o,
|
||||
|
||||
@@ -3343,7 +3343,6 @@ async fn test_duckdb_ffi(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
/// Test that flow substeps with tags that are not available for the workspace fail.
|
||||
/// This validates that `check_tag_available_for_workspace_internal` is properly called
|
||||
/// when pushing jobs from worker_flow.
|
||||
#[cfg(feature = "deno_core")]
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn test_flow_substep_tag_availability_check(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
use windmill_common::worker::{
|
||||
@@ -3385,6 +3384,7 @@ async fn test_flow_substep_tag_availability_check(db: Pool<Postgres>) -> anyhow:
|
||||
|
||||
let result =
|
||||
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
|
||||
.email("test2@windmill.dev")
|
||||
.run_until_complete(&db, false, server.addr.port())
|
||||
.await;
|
||||
|
||||
|
||||
@@ -5,6 +5,31 @@ use crate::{
|
||||
use serde::Serialize;
|
||||
use tokio::time::Instant;
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct PoolStats {
|
||||
pub peak_active_conns: u32,
|
||||
pub pool_saturation_histogram: Vec<u32>,
|
||||
}
|
||||
|
||||
impl PoolStats {
|
||||
pub fn new(pool_size: u32) -> Self {
|
||||
PoolStats {
|
||||
peak_active_conns: 0,
|
||||
pool_saturation_histogram: vec![0; pool_size as usize + 1],
|
||||
}
|
||||
}
|
||||
|
||||
pub fn sample(&mut self, pool_size: u32, num_idle: u32) {
|
||||
let active = pool_size.saturating_sub(num_idle);
|
||||
if active > self.peak_active_conns {
|
||||
self.peak_active_conns = active;
|
||||
}
|
||||
if let Some(bucket) = self.pool_saturation_histogram.get_mut(active as usize) {
|
||||
*bucket += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct BenchmarkInfo {
|
||||
#[serde(skip)]
|
||||
@@ -14,6 +39,7 @@ pub struct BenchmarkInfo {
|
||||
timings: Vec<BenchmarkIter>,
|
||||
pub iter_durations: Vec<u64>,
|
||||
pub total_duration: Option<u64>,
|
||||
pub pool_stats: Option<PoolStats>,
|
||||
}
|
||||
|
||||
impl BenchmarkInfo {
|
||||
@@ -24,6 +50,17 @@ impl BenchmarkInfo {
|
||||
start: Instant::now(),
|
||||
iter_durations: vec![],
|
||||
total_duration: None,
|
||||
pool_stats: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn init_pool_stats(&mut self, pool_size: u32) {
|
||||
self.pool_stats = Some(PoolStats::new(pool_size));
|
||||
}
|
||||
|
||||
pub fn sample_pool(&mut self, pool_size: u32, num_idle: u32) {
|
||||
if let Some(stats) = self.pool_stats.as_mut() {
|
||||
stats.sample(pool_size, num_idle);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,8 +77,11 @@ impl BenchmarkInfo {
|
||||
let total_duration = self.start.elapsed().as_millis() as u64;
|
||||
self.total_duration = Some(total_duration as u64);
|
||||
|
||||
let pool_info = self.pool_stats.as_ref().map_or(String::new(), |ps| {
|
||||
format!(", peak active conns: {}", ps.peak_active_conns)
|
||||
});
|
||||
println!(
|
||||
"Writing benchmark {path}, duration of benchmark: {total_duration}ms and RPS: {}",
|
||||
"Writing benchmark {path}, duration of benchmark: {total_duration}ms and RPS: {}{pool_info}",
|
||||
self.iters as f64 / total_duration as f64 * 1000.0
|
||||
);
|
||||
write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling");
|
||||
@@ -220,6 +260,7 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (4)"));
|
||||
}
|
||||
"none" => {}
|
||||
_ => {
|
||||
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
|
||||
None::<i64>,
|
||||
|
||||
@@ -1184,7 +1184,19 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
|
||||
.concurrent_limit
|
||||
.is_some()
|
||||
{
|
||||
let concurrency_key = concurrency_key(db, &completed_job.id).await?;
|
||||
let concurrency_key = sqlx::query_scalar!(
|
||||
"SELECT key FROM concurrency_key WHERE job_id = $1",
|
||||
&completed_job.id
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.warn_after_seconds(10)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
Error::internal_err(format!(
|
||||
"Could not get concurrency key for job {}: {e:#}",
|
||||
completed_job.id
|
||||
))
|
||||
})?;
|
||||
if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_none() {
|
||||
tracing::warn!("Concurrency limit is disabled, skipping");
|
||||
} else {
|
||||
@@ -1221,18 +1233,6 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
|
||||
tracing::debug!("decremented concurrency counter");
|
||||
}
|
||||
|
||||
sqlx::query!("DELETE FROM job_perms WHERE job_id = $1", job_id)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(10)
|
||||
.await?;
|
||||
|
||||
if !success || has_stream {
|
||||
sqlx::query!("DELETE FROM job_result_stream_v2 WHERE job_id = $1", job_id)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(10)
|
||||
.await?;
|
||||
}
|
||||
|
||||
tx.commit().warn_after_seconds(10).await?;
|
||||
|
||||
tracing::info!(
|
||||
@@ -5386,7 +5386,7 @@ async fn push_inner<'c, 'd>(
|
||||
&runnable_path,
|
||||
job_kind,
|
||||
concurrency_settings.concurrency_key.clone(),
|
||||
&mut tx,
|
||||
&mut *tx,
|
||||
job_id,
|
||||
)
|
||||
.await?;
|
||||
@@ -5709,7 +5709,7 @@ pub async fn insert_concurrency_key<'d, 'c>(
|
||||
script_path: &Option<String>,
|
||||
job_kind: JobKind,
|
||||
custom_concurrency_key: Option<String>,
|
||||
tx: &mut Transaction<'c, Postgres>,
|
||||
db: impl PgExecutor<'c>,
|
||||
job_id: Uuid,
|
||||
) -> Result<(), Error> {
|
||||
let concurrency_key = custom_concurrency_key
|
||||
@@ -5737,7 +5737,7 @@ pub async fn insert_concurrency_key<'d, 'c>(
|
||||
));
|
||||
sqlx::query!(
|
||||
"WITH inserted_concurrency_counter AS (
|
||||
INSERT INTO concurrency_counter (concurrency_id, job_uuids)
|
||||
INSERT INTO concurrency_counter (concurrency_id, job_uuids)
|
||||
VALUES ($1, '{}'::jsonb)
|
||||
ON CONFLICT DO NOTHING
|
||||
)
|
||||
@@ -5745,7 +5745,7 @@ pub async fn insert_concurrency_key<'d, 'c>(
|
||||
concurrency_key,
|
||||
job_id,
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.execute(db)
|
||||
.warn_after_seconds(3)
|
||||
.await
|
||||
.map_err(|e| Error::internal_err(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e:#}")))?;
|
||||
|
||||
@@ -1618,6 +1618,11 @@ pub async fn run_worker(
|
||||
#[cfg(feature = "benchmark")]
|
||||
let mut infos = BenchmarkInfo::new();
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
if let Some(db) = conn.as_sql() {
|
||||
infos.init_pool_stats(db.size());
|
||||
}
|
||||
|
||||
let vacuum_shift = rand::rng().random_range(0..VACUUM_PERIOD);
|
||||
|
||||
IS_READY.store(true, Ordering::Relaxed);
|
||||
@@ -2067,6 +2072,9 @@ pub async fn run_worker(
|
||||
#[cfg(feature = "benchmark")]
|
||||
{
|
||||
add_time!(bench, "sent to dedicated worker");
|
||||
if let Some(db) = conn.as_sql() {
|
||||
infos.sample_pool(db.size(), db.num_idle() as u32);
|
||||
}
|
||||
infos.add_iter(bench, true);
|
||||
}
|
||||
|
||||
@@ -2137,6 +2145,9 @@ pub async fn run_worker(
|
||||
#[cfg(feature = "benchmark")]
|
||||
{
|
||||
add_time!(bench, "sent to flow runner");
|
||||
if let Some(db) = conn.as_sql() {
|
||||
infos.sample_pool(db.size(), db.num_idle() as u32);
|
||||
}
|
||||
infos.add_iter(bench, true);
|
||||
}
|
||||
|
||||
@@ -2437,6 +2448,9 @@ pub async fn run_worker(
|
||||
{
|
||||
if started {
|
||||
add_time!(bench, "job processed");
|
||||
if let Some(db) = conn.as_sql() {
|
||||
infos.sample_pool(db.size(), db.num_idle() as u32);
|
||||
}
|
||||
infos.add_iter(bench, true);
|
||||
}
|
||||
}
|
||||
@@ -2465,6 +2479,9 @@ pub async fn run_worker(
|
||||
#[cfg(feature = "benchmark")]
|
||||
{
|
||||
add_time!(bench, "sleep because empty job queue");
|
||||
if let Some(db) = conn.as_sql() {
|
||||
infos.sample_pool(db.size(), db.num_idle() as u32);
|
||||
}
|
||||
infos.add_iter(bench, false);
|
||||
}
|
||||
#[cfg(feature = "prometheus")]
|
||||
|
||||
@@ -1365,35 +1365,31 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
let args_hm = args.unwrap_or_default().0;
|
||||
let args = PushArgs::from(&args_hm);
|
||||
if let Some(ck) = concurrency_key {
|
||||
let mut tx = db.begin().await?;
|
||||
insert_concurrency_key(
|
||||
&flow_job.workspace_id,
|
||||
&args,
|
||||
&flow_job.runnable_path,
|
||||
JobKind::Flow,
|
||||
Some(ck),
|
||||
&mut tx,
|
||||
db,
|
||||
flow,
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
}
|
||||
if let Some(t) = tag {
|
||||
tag = Some(interpolate_args(t, &args, &flow_job.workspace_id));
|
||||
}
|
||||
} else if concurrent_limit.is_some() {
|
||||
let mut tx = db.begin().await?;
|
||||
insert_concurrency_key(
|
||||
&flow_job.workspace_id,
|
||||
&PushArgs::from(&HashMap::new()),
|
||||
&flow_job.runnable_path,
|
||||
JobKind::Flow,
|
||||
concurrency_key,
|
||||
&mut tx,
|
||||
db,
|
||||
flow,
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
}
|
||||
|
||||
// let tag = tag_and_concurrency_key.and_then(|tc| tc.tag.map(|t| interpolate_args(t.clone(), &args, &workspace_id)));
|
||||
@@ -3257,7 +3253,7 @@ async fn push_next_flow_job(
|
||||
"UPDATE v2_job_runtime SET ping = now() WHERE id = $1 AND ping < now()",
|
||||
flow_job.id,
|
||||
)
|
||||
.execute(db)
|
||||
.execute(&mut *tx)
|
||||
.warn_after_seconds(3)
|
||||
.await?;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user