mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-24 16:01:42 +00:00
feat: Job caching now works with S3 files (#2934)
This commit is contained in:
Guillaume Bouvignies
committed by
GitHub
GitHub
parent
94b3d9f9c7
commit
08d24fe37b
Generated
+67
-33
@@ -1499,9 +1499,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "chrono-tz"
|
||||
version = "0.8.4"
|
||||
version = "0.8.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e23185c0e21df6ed832a12e2bda87c7d1def6842881fb634a8511ced741b0d76"
|
||||
checksum = "91d7b79e99bfaa0d47da0687c43aa3b7381938a62ad3a6498599039321f660b7"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"chrono-tz-build",
|
||||
@@ -1551,9 +1551,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "clap"
|
||||
version = "4.4.11"
|
||||
version = "4.4.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bfaff671f6b22ca62406885ece523383b9b64022e341e53e009a62ebc47a45f2"
|
||||
checksum = "dcfab8ba68f3668e89f6ff60f5b205cea56aa7b769451a59f34b8682f51c056d"
|
||||
dependencies = [
|
||||
"clap_builder",
|
||||
"clap_derive",
|
||||
@@ -1561,9 +1561,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "clap_builder"
|
||||
version = "4.4.11"
|
||||
version = "4.4.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a216b506622bb1d316cd51328dce24e07bdff4a6128a47c7e7fad11878d5adbb"
|
||||
checksum = "fb7fb5e4e979aec3be7791562fcba452f94ad85e954da024396433e0e25a79e9"
|
||||
dependencies = [
|
||||
"anstream",
|
||||
"anstyle",
|
||||
@@ -3742,6 +3742,15 @@ dependencies = [
|
||||
"syn 2.0.43",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "itertools"
|
||||
version = "0.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "284f18f85651fe11e8a991b2adb42cb078325c996ed026d994719efcfca1d54b"
|
||||
dependencies = [
|
||||
"either",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "itertools"
|
||||
version = "0.10.5"
|
||||
@@ -3813,6 +3822,15 @@ dependencies = [
|
||||
"simple_asn1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "keccak"
|
||||
version = "0.1.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8f6d5ed8676d904364de097082f4e7d240b571b67989ced0240f08b7f966f940"
|
||||
dependencies = [
|
||||
"cpufeatures",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "keyed_priority_queue"
|
||||
version = "0.4.2"
|
||||
@@ -4104,9 +4122,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mail-send"
|
||||
version = "0.4.3"
|
||||
version = "0.4.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e373c9ec9178397af6ea2964568e49e1d486b2d80d17b06aa19da75566416422"
|
||||
checksum = "4e38e7453c5cb25cbeaa251e480c74592ec121360a6bcb822928926bf69e3d1e"
|
||||
dependencies = [
|
||||
"base64 0.21.5",
|
||||
"gethostname",
|
||||
@@ -4121,9 +4139,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "malachite"
|
||||
version = "0.4.4"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "220cb36c52aa6eff45559df497abe0e2a4c1209f92279a746a399f622d7b95c7"
|
||||
checksum = "f6cf7f4730c30071ba374fac86ad35b1cb7a0716f774737768667ea3fa1828e3"
|
||||
dependencies = [
|
||||
"malachite-base",
|
||||
"malachite-nz",
|
||||
@@ -4132,19 +4150,22 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "malachite-base"
|
||||
version = "0.4.4"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6538136c5daf04126d6be4899f7fe4879b7f8de896dd1b4210fe6de5b94f2555"
|
||||
checksum = "2b06bfa98a4b4802af5a4263b4ad4660e28e51e8490f6354eb9336c70767e1c5"
|
||||
dependencies = [
|
||||
"itertools 0.11.0",
|
||||
"itertools 0.9.0",
|
||||
"rand 0.7.3",
|
||||
"rand_chacha 0.2.2",
|
||||
"ryu",
|
||||
"sha3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "malachite-bigint"
|
||||
version = "0.1.1"
|
||||
version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "76c3eca3b5df299486144c8423c45c24bdf9e82e2452c8a1eeda547c4d8b5d41"
|
||||
checksum = "8a5110aee54537b0cef214efbebdd7df79b7408db8eef4f6a4b6db9d0d8fc01b"
|
||||
dependencies = [
|
||||
"derive_more",
|
||||
"malachite",
|
||||
@@ -4155,22 +4176,22 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "malachite-nz"
|
||||
version = "0.4.4"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5f0b05577b7a3f09433106460b10304f97fc572f0baabf6640e6cb1e23f5fc52"
|
||||
checksum = "c89e21c64b7af5be3dc8cef16f786243faf59459fe4ba93b44efdeb264e5ade4"
|
||||
dependencies = [
|
||||
"embed-doc-image",
|
||||
"itertools 0.11.0",
|
||||
"itertools 0.9.0",
|
||||
"malachite-base",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "malachite-q"
|
||||
version = "0.4.4"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a1cfdb4016292e6acd832eaee261175f3af8bbee62afeefe4420ebce4c440cb5"
|
||||
checksum = "3755e541d5134b5016594c9043094172c4dda9259b3ce824a7b8101941850360"
|
||||
dependencies = [
|
||||
"itertools 0.11.0",
|
||||
"itertools 0.9.0",
|
||||
"malachite-base",
|
||||
"malachite-nz",
|
||||
]
|
||||
@@ -4219,9 +4240,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "memchr"
|
||||
version = "2.6.4"
|
||||
version = "2.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f665ee40bc4a3c5590afb1e9677db74a508659dfd71e126420da8274909a0167"
|
||||
checksum = "523dc4f511e55ab87b694dc30d0f820d60906ef06413f93d4d7a1385599cc149"
|
||||
|
||||
[[package]]
|
||||
name = "memmap2"
|
||||
@@ -7000,6 +7021,18 @@ dependencies = [
|
||||
"digest 0.10.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sha3"
|
||||
version = "0.9.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f81199417d4e5de3f04b1e871023acea7389672c4135918f05aa9cbf2f2fa809"
|
||||
dependencies = [
|
||||
"block-buffer 0.9.0",
|
||||
"digest 0.9.0",
|
||||
"keccak",
|
||||
"opaque-debug",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sharded-slab"
|
||||
version = "0.1.7"
|
||||
@@ -7116,9 +7149,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "smtp-proto"
|
||||
version = "0.1.1"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d4b756ac662e92a0e5b360349bea5f0b0784d4be4541eff2972049dfdfd7f862"
|
||||
checksum = "0f7278d054ed6844babe3d760d03ccae24e26207c3c27b68d8385a03d3324619"
|
||||
|
||||
[[package]]
|
||||
name = "snafu"
|
||||
@@ -8156,15 +8189,15 @@ checksum = "cfb5fa503293557c5158bd215fdc225695e567a77e453f5d4452a50a193969bd"
|
||||
|
||||
[[package]]
|
||||
name = "tempfile"
|
||||
version = "3.8.1"
|
||||
version = "3.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7ef1adac450ad7f4b3c28589471ade84f25f731a7a0fe30d71dfa9f60fd808e5"
|
||||
checksum = "01ce4141aa927a6d1bd34a041795abd0db1cccba5d5f24b009f694bdf3a1f3fa"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"fastrand 2.0.1",
|
||||
"redox_syscall 0.4.1",
|
||||
"rustix 0.38.28",
|
||||
"windows-sys 0.48.0",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -9452,7 +9485,6 @@ dependencies = [
|
||||
"async-recursion",
|
||||
"async-stripe",
|
||||
"async_zip",
|
||||
"aws-config",
|
||||
"aws-sdk-s3",
|
||||
"axum",
|
||||
"base64 0.21.5",
|
||||
@@ -9550,6 +9582,8 @@ name = "windmill-common"
|
||||
version = "1.232.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"aws-config",
|
||||
"aws-sdk-s3",
|
||||
"axum",
|
||||
"chrono",
|
||||
"cron",
|
||||
@@ -9996,9 +10030,9 @@ checksum = "dff9641d1cd4be8d1a070daf9e3773c5f67e78b4d9d42263020c057706765c04"
|
||||
|
||||
[[package]]
|
||||
name = "winnow"
|
||||
version = "0.5.30"
|
||||
version = "0.5.31"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9b5c3db89721d50d0e2a673f5043fc4722f76dcc352d7b1ab8b8288bed4ed2c5"
|
||||
checksum = "97a4882e6b134d6c28953a387571f1acdd3496830d5e36c5e3a1075580ea641c"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
@@ -10024,9 +10058,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "xattr"
|
||||
version = "1.1.3"
|
||||
version = "1.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a7dae5072fe1f8db8f8d29059189ac175196e410e40ba42d5d4684ae2f750995"
|
||||
checksum = "914566e6413e7fa959cc394fb30e563ba80f3541fbd40816d4c05a0fc3f2a0f1"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"linux-raw-sys 0.4.12",
|
||||
|
||||
+4
-1
@@ -208,4 +208,7 @@ candle-transformers = "0.3.0"
|
||||
candle-nn = "0.3.0"
|
||||
tiberius = { version = "0.12.2", default-features = false, features = ["rustls", "tds73", "chrono"] }
|
||||
aws-sdk-s3 = "1.5.0"
|
||||
aws-config = "1.0.3"
|
||||
aws-config = "1.0.3"
|
||||
polars = { version = "0.35.2", features = ["lazy", "parquet", "aws", "csv", "dtype-full"] }
|
||||
polars-io = { version = "0.35.2", features = ["csv"] }
|
||||
object_store = { version = "0.8.0", features = ["aws"] }
|
||||
|
||||
@@ -84,7 +84,6 @@ candle-core.workspace = true
|
||||
candle-transformers.workspace = true
|
||||
candle-nn.workspace = true
|
||||
aws-sdk-s3.workspace = true
|
||||
aws-config.workspace = true
|
||||
polars = { version = "0.35.2", features = ["lazy", "parquet", "aws", "csv", "dtype-full"] }
|
||||
polars-io = { version = "0.35.2", features = ["csv"] }
|
||||
object_store = { version = "0.8.0", features = ["aws"] }
|
||||
polars.workspace = true
|
||||
polars-io.workspace = true
|
||||
object_store.workspace = true
|
||||
|
||||
@@ -1,12 +1,8 @@
|
||||
use std::{cmp, time::Duration};
|
||||
|
||||
use crate::{
|
||||
db::DB, resources::get_resource_value_interpolated_internal, users::Tokened,
|
||||
workspaces::LargeFileStorage,
|
||||
};
|
||||
use crate::{db::DB, resources::get_resource_value_interpolated_internal, users::Tokened};
|
||||
use anyhow::Context;
|
||||
use aws_sdk_s3::{
|
||||
config::{BehaviorVersion, Credentials, Region},
|
||||
presigning::PresigningConfig,
|
||||
primitives::ByteStream,
|
||||
types::{CompletedMultipartUpload, CompletedPart},
|
||||
@@ -32,7 +28,11 @@ use polars::{
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tower_http::cors::{Any, CorsLayer};
|
||||
use windmill_common::{db::UserDB, error};
|
||||
use windmill_common::{
|
||||
db::UserDB,
|
||||
error,
|
||||
s3_helpers::{build_s3_client, render_endpoint, LargeFileStorage, S3Resource},
|
||||
};
|
||||
|
||||
use crate::db::ApiAuthed;
|
||||
|
||||
@@ -87,23 +87,6 @@ pub fn workspaced_service() -> Router {
|
||||
)
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, Serialize, Clone)]
|
||||
struct S3Resource {
|
||||
#[serde(rename = "bucket")]
|
||||
bucket: String,
|
||||
region: String,
|
||||
#[serde(rename = "endPoint")]
|
||||
endpoint: String,
|
||||
#[serde(rename = "useSSL")]
|
||||
use_ssl: bool,
|
||||
#[serde(rename = "accessKey")]
|
||||
access_key: Option<String>,
|
||||
#[serde(rename = "secretKey")]
|
||||
secret_key: Option<String>,
|
||||
#[serde(rename = "pathStyle")]
|
||||
path_style: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct DuckdbConnectionSettingsQuery {
|
||||
s3_resource: S3Resource,
|
||||
@@ -346,7 +329,10 @@ async fn test_connection(
|
||||
.max_keys(1)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| error::Error::InternalErr(err.to_string()))?;
|
||||
.map_err(|err| {
|
||||
tracing::error!("Error testing connection to S3 bucket: {:?}", err);
|
||||
error::Error::InternalErr(err.to_string())
|
||||
})?;
|
||||
return Ok(Json(()));
|
||||
}
|
||||
|
||||
@@ -911,6 +897,11 @@ async fn multipart_upload_s3_file(
|
||||
}));
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct S3Object {
|
||||
pub s3: String,
|
||||
}
|
||||
|
||||
async fn get_workspace_s3_resource<'c>(
|
||||
authed: &ApiAuthed,
|
||||
user_db: &UserDB,
|
||||
@@ -983,30 +974,6 @@ async fn get_s3_resource<'c>(
|
||||
return Ok(Some(s3_resource));
|
||||
}
|
||||
|
||||
fn build_s3_client(s3_resource_ref: &S3Resource) -> aws_sdk_s3::Client {
|
||||
let s3_resource = s3_resource_ref.clone();
|
||||
|
||||
let endpoint = render_endpoint(&s3_resource);
|
||||
let mut s3_config_builder = aws_sdk_s3::Config::builder()
|
||||
.endpoint_url(endpoint)
|
||||
.behavior_version(BehaviorVersion::latest())
|
||||
.region(Region::new(s3_resource.region));
|
||||
if s3_resource.access_key.is_some() {
|
||||
s3_config_builder = s3_config_builder.credentials_provider(Credentials::new(
|
||||
s3_resource.access_key.unwrap_or_default(),
|
||||
s3_resource.secret_key.unwrap_or_default(),
|
||||
None,
|
||||
None,
|
||||
"s3_storage",
|
||||
));
|
||||
}
|
||||
if s3_resource.path_style {
|
||||
s3_config_builder = s3_config_builder.force_path_style(true);
|
||||
}
|
||||
let s3_config = s3_config_builder.build();
|
||||
return aws_sdk_s3::Client::from_conf(s3_config);
|
||||
}
|
||||
|
||||
fn build_polars_s3_config(s3_resource_ref: &S3Resource) -> CloudOptions {
|
||||
let s3_resource = s3_resource_ref.to_owned();
|
||||
let mut s3_configs: Vec<(AmazonS3ConfigKey, String)> = vec![
|
||||
@@ -1034,16 +1001,6 @@ fn build_polars_s3_config(s3_resource_ref: &S3Resource) -> CloudOptions {
|
||||
return CloudOptions::default().with_aws(s3_configs);
|
||||
}
|
||||
|
||||
fn render_endpoint(s3_resource: &S3Resource) -> String {
|
||||
if s3_resource.endpoint.starts_with("http://") || s3_resource.endpoint.starts_with("https://") {
|
||||
s3_resource.endpoint.clone()
|
||||
} else if s3_resource.use_ssl {
|
||||
format!("https://{}", s3_resource.endpoint)
|
||||
} else {
|
||||
format!("http://{}", s3_resource.endpoint)
|
||||
}
|
||||
}
|
||||
|
||||
async fn read_s3_text_object_head(
|
||||
s3_client: &aws_sdk_s3::Client,
|
||||
s3_bucket: &str,
|
||||
|
||||
@@ -40,6 +40,7 @@ use stripe::CustomerId;
|
||||
use uuid::Uuid;
|
||||
use windmill_audit::{audit_log, ActionKind};
|
||||
use windmill_common::db::UserDB;
|
||||
use windmill_common::s3_helpers::LargeFileStorage;
|
||||
use windmill_common::schedule::Schedule;
|
||||
use windmill_common::users::username_to_permissioned_as;
|
||||
use windmill_common::worker::CLOUD_HOSTED;
|
||||
@@ -281,18 +282,6 @@ pub struct EditErrorHandler {
|
||||
pub error_handler_muted_on_cancel: Option<bool>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum LargeFileStorage {
|
||||
S3Storage(S3Storage),
|
||||
// TODO: Add a filesystem type here in the future if needed
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug)]
|
||||
pub struct S3Storage {
|
||||
pub s3_resource_path: String,
|
||||
}
|
||||
|
||||
async fn list_pending_invites(
|
||||
authed: ApiAuthed,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
|
||||
@@ -48,4 +48,6 @@ itertools.workspace = true
|
||||
regex.workspace = true
|
||||
git-version.workspace = true
|
||||
cron.workspace = true
|
||||
tracing-loki = { version = "^0", optional = true }
|
||||
tracing-loki = { version = "^0", optional = true }
|
||||
aws-sdk-s3.workspace = true
|
||||
aws-config.workspace = true
|
||||
@@ -26,6 +26,7 @@ pub mod global_settings;
|
||||
pub mod jobs;
|
||||
pub mod more_serde;
|
||||
pub mod oauth2;
|
||||
pub mod s3_helpers;
|
||||
pub mod schedule;
|
||||
pub mod scripts;
|
||||
pub mod server;
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
use aws_config::{BehaviorVersion, Region};
|
||||
use aws_sdk_s3::config::Credentials;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum LargeFileStorage {
|
||||
S3Storage(S3Storage),
|
||||
// TODO: Add a filesystem type here in the future if needed
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug)]
|
||||
pub struct S3Storage {
|
||||
pub s3_resource_path: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, Serialize, Clone)]
|
||||
pub struct S3Resource {
|
||||
#[serde(rename = "bucket")]
|
||||
pub bucket: String,
|
||||
pub region: String,
|
||||
#[serde(rename = "endPoint")]
|
||||
pub endpoint: String,
|
||||
#[serde(rename = "useSSL")]
|
||||
pub use_ssl: bool,
|
||||
#[serde(rename = "accessKey")]
|
||||
pub access_key: Option<String>,
|
||||
#[serde(rename = "secretKey")]
|
||||
pub secret_key: Option<String>,
|
||||
#[serde(rename = "pathStyle")]
|
||||
pub path_style: bool,
|
||||
pub port: Option<u16>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Clone)]
|
||||
pub struct S3Object {
|
||||
pub s3: String,
|
||||
}
|
||||
|
||||
pub async fn get_etag_or_empty(s3_resource: &S3Resource, s3_object: S3Object) -> Option<String> {
|
||||
let s3_client = build_s3_client(s3_resource);
|
||||
|
||||
let s3_bucket = s3_resource.bucket.clone();
|
||||
let s3_key = s3_object.s3;
|
||||
let s3_object_metadata = s3_client
|
||||
.head_object()
|
||||
.bucket(s3_bucket)
|
||||
.key(s3_key)
|
||||
.send()
|
||||
.await;
|
||||
s3_object_metadata
|
||||
.ok()
|
||||
.map(|res| res.e_tag().map(|et| et.to_string()))
|
||||
.flatten()
|
||||
}
|
||||
|
||||
pub fn render_endpoint(s3_resource: &S3Resource) -> String {
|
||||
let url_with_prefix = if s3_resource.endpoint.starts_with("http://")
|
||||
|| s3_resource.endpoint.starts_with("https://")
|
||||
{
|
||||
s3_resource.endpoint.clone()
|
||||
} else if s3_resource.use_ssl {
|
||||
format!("https://{}", s3_resource.endpoint)
|
||||
} else {
|
||||
format!("http://{}", s3_resource.endpoint)
|
||||
};
|
||||
if s3_resource.port.is_some() {
|
||||
format!("{}:{}", url_with_prefix, s3_resource.port.unwrap())
|
||||
} else {
|
||||
url_with_prefix
|
||||
}
|
||||
}
|
||||
|
||||
pub fn build_s3_client(s3_resource_ref: &S3Resource) -> aws_sdk_s3::Client {
|
||||
let s3_resource = s3_resource_ref.clone();
|
||||
|
||||
let endpoint = render_endpoint(&s3_resource);
|
||||
let mut s3_config_builder = aws_sdk_s3::Config::builder()
|
||||
.endpoint_url(endpoint)
|
||||
.behavior_version(BehaviorVersion::latest())
|
||||
.region(Region::new(s3_resource.region));
|
||||
if s3_resource.access_key.is_some() {
|
||||
s3_config_builder = s3_config_builder.credentials_provider(Credentials::new(
|
||||
s3_resource.access_key.unwrap_or_default(),
|
||||
s3_resource.secret_key.unwrap_or_default(),
|
||||
None,
|
||||
None,
|
||||
"s3_storage",
|
||||
));
|
||||
}
|
||||
if s3_resource.path_style {
|
||||
s3_config_builder = s3_config_builder.force_path_style(true);
|
||||
}
|
||||
let s3_config = s3_config_builder.build();
|
||||
return aws_sdk_s3::Client::from_conf(s3_config);
|
||||
}
|
||||
@@ -3,12 +3,13 @@ use itertools::Itertools;
|
||||
use nix::sys::signal::{self, Signal};
|
||||
use nix::unistd::Pid;
|
||||
use regex::Regex;
|
||||
use serde::Serialize;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::value::RawValue;
|
||||
use serde_json::{json, Value};
|
||||
use sqlx::{Pool, Postgres};
|
||||
use tokio::process::Command;
|
||||
use tokio::{fs::File, io::AsyncReadExt};
|
||||
use windmill_common::s3_helpers::{get_etag_or_empty, LargeFileStorage, S3Object, S3Resource};
|
||||
use windmill_common::worker::{CLOUD_HOSTED, WORKER_CONFIG};
|
||||
use windmill_common::{
|
||||
error::{self, Error},
|
||||
@@ -31,7 +32,7 @@ use std::{
|
||||
|
||||
use tracing::{trace_span, Instrument};
|
||||
use uuid::Uuid;
|
||||
use windmill_common::variables;
|
||||
use windmill_common::{variables, DB};
|
||||
|
||||
use tokio::{
|
||||
io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
|
||||
@@ -841,13 +842,26 @@ fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize, remove_x00:
|
||||
}
|
||||
}
|
||||
|
||||
pub fn hash_args(v: &Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>) -> String {
|
||||
pub async fn hash_args(
|
||||
db: &DB,
|
||||
client: &AuthedClient,
|
||||
workspace_id: &str,
|
||||
job_id: &Uuid,
|
||||
v: &Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
|
||||
) -> String {
|
||||
if let Some(vs) = v {
|
||||
let mut dh = DefaultHasher::new();
|
||||
let hm = &vs.0;
|
||||
for k in hm.keys().sorted() {
|
||||
k.hash(&mut dh);
|
||||
hm.get(k).unwrap().get().hash(&mut dh);
|
||||
let arg_value = hm.get(k).unwrap();
|
||||
let arg_additions =
|
||||
arg_value_hash_additions(db, client, workspace_id, job_id, hm.get(k).unwrap())
|
||||
.await;
|
||||
arg_value.get().hash(&mut dh);
|
||||
for (_, arg_addition) in arg_additions {
|
||||
arg_addition.hash(&mut dh);
|
||||
}
|
||||
}
|
||||
hex::encode(dh.finish().to_be_bytes())
|
||||
} else {
|
||||
@@ -855,21 +869,152 @@ pub fn hash_args(v: &Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct StoreCachedResource<'a> {
|
||||
expire: i64,
|
||||
value: &'a RawValue,
|
||||
async fn get_workspace_s3_resource_path(
|
||||
db: &DB,
|
||||
client: &AuthedClient,
|
||||
workspace_id: &str,
|
||||
job_id: &Uuid,
|
||||
) -> Option<S3Resource> {
|
||||
let raw_lfs_opt = sqlx::query_scalar!(
|
||||
"SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1",
|
||||
workspace_id
|
||||
)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.map(|val| serde_json::from_value::<LargeFileStorage>(val).ok())
|
||||
.flatten();
|
||||
|
||||
if let Some(LargeFileStorage::S3Storage(s3_storage)) = raw_lfs_opt {
|
||||
let resource_path = s3_storage.s3_resource_path.trim_start_matches("$res:");
|
||||
client
|
||||
.get_resource_value_interpolated::<S3Resource>(&resource_path, Some(job_id.to_string()))
|
||||
.await
|
||||
.ok()
|
||||
} else {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn save_in_cache<'a>(
|
||||
async fn arg_value_hash_additions(
|
||||
db: &DB,
|
||||
client: &AuthedClient,
|
||||
workspace_id: &str,
|
||||
job_id: &Uuid,
|
||||
raw_value: &Box<RawValue>,
|
||||
) -> HashMap<String, String> {
|
||||
let mut result: HashMap<String, String> = HashMap::new();
|
||||
let parsed_s3_values: &mut Vec<S3Object> = &mut Vec::new();
|
||||
extract_all_s3_object_from_raw_value(raw_value, parsed_s3_values);
|
||||
if parsed_s3_values.is_empty() {
|
||||
// no s3 object, nothing to return
|
||||
return result;
|
||||
}
|
||||
|
||||
let s3_resource_opt = get_workspace_s3_resource_path(db, client, workspace_id, job_id).await;
|
||||
if let Some(s3_resource) = s3_resource_opt {
|
||||
for s3_object in parsed_s3_values {
|
||||
let etag = get_etag_or_empty(&s3_resource, s3_object.clone()).await;
|
||||
tracing::warn!("Enriching s3 arg value with etag: {:?}", etag);
|
||||
result.insert(s3_object.s3.clone(), etag.unwrap_or_default()); // TODO: maybe inject a random value to invalidate the cache?
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
fn extract_all_s3_object_from_raw_value(raw_value: &Box<RawValue>, result: &mut Vec<S3Object>) {
|
||||
let parsed_value = serde_json::from_str::<S3Object>(raw_value.get());
|
||||
if let Ok(parsed_value) = parsed_value {
|
||||
result.push(parsed_value);
|
||||
} else {
|
||||
let parsed_value = serde_json::from_str::<HashMap<String, Box<RawValue>>>(raw_value.get());
|
||||
if let Ok(parsed_value) = parsed_value {
|
||||
for (_, v) in parsed_value {
|
||||
extract_all_s3_object_from_raw_value(&v, result);
|
||||
}
|
||||
} else {
|
||||
let parsed_value = serde_json::from_str::<Vec<Box<RawValue>>>(raw_value.get());
|
||||
if let Ok(parsed_value) = parsed_value {
|
||||
for v in parsed_value {
|
||||
extract_all_s3_object_from_raw_value(&v, result);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize)]
|
||||
struct CachedResource {
|
||||
expire: i64,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
s3_etags: Option<HashMap<String, String>>,
|
||||
value: Box<RawValue>,
|
||||
}
|
||||
|
||||
pub async fn get_cached_resource_value_if_valid(
|
||||
db: &DB,
|
||||
client: &AuthedClient,
|
||||
job_id: &Uuid,
|
||||
workspace_id: &str,
|
||||
cached_res_path: &str,
|
||||
) -> Option<Box<RawValue>> {
|
||||
let resource_opt = client
|
||||
.get_resource_value::<CachedResource>(cached_res_path)
|
||||
.await
|
||||
.ok();
|
||||
if let Some(cached_resource) = resource_opt {
|
||||
if cached_resource.expire <= chrono::Utc::now().timestamp() {
|
||||
// cache expired
|
||||
return None;
|
||||
}
|
||||
let s3_etags = cached_resource.s3_etags.unwrap_or_default();
|
||||
let s3_resource_opt: Option<S3Resource> = if s3_etags.is_empty() {
|
||||
None
|
||||
} else {
|
||||
get_workspace_s3_resource_path(db, &client, workspace_id, job_id).await
|
||||
};
|
||||
if !s3_etags.is_empty() && s3_resource_opt.is_none() {
|
||||
tracing::warn!("Cached result references s3 files that are not retrievable anymore because the workspace S3 resource can't be fetched. Cache will be invalidated");
|
||||
return None;
|
||||
}
|
||||
for (s3_file_key, s3_file_etag) in s3_etags {
|
||||
if let Some(s3_resource) = s3_resource_opt.clone() {
|
||||
let etag =
|
||||
get_etag_or_empty(&s3_resource, S3Object { s3: s3_file_key.clone() }).await;
|
||||
if etag.is_none() || etag.clone().unwrap() != s3_file_etag {
|
||||
tracing::warn!("S3 file etag for '{}' has changed. Value from cache is {:?} while current value from S3 is {:?}. Cache will be invalidated", s3_file_key.clone(), s3_file_etag, etag);
|
||||
return None;
|
||||
}
|
||||
}
|
||||
}
|
||||
return Some(cached_resource.value);
|
||||
}
|
||||
return None;
|
||||
}
|
||||
|
||||
pub async fn save_in_cache(
|
||||
db: &Pool<Postgres>,
|
||||
client: &AuthedClient,
|
||||
job: &QueuedJob,
|
||||
cached_path: String,
|
||||
r: &'a RawValue,
|
||||
r: &Box<RawValue>,
|
||||
) {
|
||||
let expire = chrono::Utc::now().timestamp() + job.cache_ttl.unwrap() as i64;
|
||||
|
||||
let store_cache_resource = StoreCachedResource { expire, value: r };
|
||||
let s3_etags =
|
||||
arg_value_hash_additions(db, client, job.workspace_id.as_str(), &job.id, r).await;
|
||||
|
||||
let store_cache_resource = CachedResource {
|
||||
expire,
|
||||
s3_etags: if s3_etags.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(s3_etags)
|
||||
},
|
||||
value: r.clone(),
|
||||
};
|
||||
let raw_json = sqlx::types::Json(store_cache_resource);
|
||||
|
||||
if let Err(e) = sqlx::query!(
|
||||
@@ -879,7 +1024,7 @@ pub async fn save_in_cache<'a>(
|
||||
DO UPDATE SET value = $3",
|
||||
job.workspace_id,
|
||||
cached_path,
|
||||
raw_json as sqlx::types::Json<StoreCachedResource>,
|
||||
raw_json as sqlx::types::Json<CachedResource>,
|
||||
"cache"
|
||||
)
|
||||
.execute(db)
|
||||
|
||||
@@ -76,7 +76,10 @@ use windmill_queue::{add_completed_job, add_completed_job_error};
|
||||
use crate::{
|
||||
bash_executor::{handle_bash_job, handle_powershell_job, ANSI_ESCAPE_RE},
|
||||
bun_executor::{gen_lockfile, get_trusted_deps, handle_bun_job},
|
||||
common::{build_args_map, hash_args, read_result, save_in_cache, write_file},
|
||||
common::{
|
||||
build_args_map, get_cached_resource_value_if_valid, hash_args, read_result, save_in_cache,
|
||||
write_file,
|
||||
},
|
||||
deno_executor::{generate_deno_lock, handle_deno_job},
|
||||
go_executor::{handle_go_job, install_go_dependencies},
|
||||
graphql_executor::do_graphql,
|
||||
@@ -2084,7 +2087,7 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
|
||||
if success {
|
||||
// println!("bef completed job{:?}", SystemTime::now());
|
||||
if let Some(cached_path) = cached_res_path {
|
||||
save_in_cache(db, &job, cached_path.to_string(), &result).await;
|
||||
save_in_cache(db, client, &job, cached_path.to_string(), &result).await;
|
||||
}
|
||||
|
||||
let timer = worker_save_completed_job_duration
|
||||
@@ -2332,12 +2335,6 @@ async fn do_nativets(
|
||||
Ok((result.0, [logs, result.1].join("\n\n")))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct CachedResource {
|
||||
expire: i64,
|
||||
value: Box<RawValue>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize, Default)]
|
||||
pub struct PreviousResult<'a> {
|
||||
#[serde(borrow)]
|
||||
@@ -2399,7 +2396,14 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
|
||||
} else {
|
||||
"none".to_string()
|
||||
};
|
||||
let args_hash = hash_args(&job.args);
|
||||
let args_hash = hash_args(
|
||||
db,
|
||||
&client.get_authed().await,
|
||||
&job.workspace_id,
|
||||
&job.id,
|
||||
&job.args,
|
||||
)
|
||||
.await;
|
||||
if job.is_flow_step {
|
||||
let flow_path = sqlx::query_scalar!(
|
||||
"SELECT script_path FROM queue WHERE id = $1",
|
||||
@@ -2424,33 +2428,32 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
|
||||
|
||||
if let Some(cached_res_path) = cached_res_path.clone() {
|
||||
let authed_client = client.get_authed().await;
|
||||
let resource = authed_client
|
||||
.get_resource_value::<CachedResource>(&cached_res_path)
|
||||
.await;
|
||||
|
||||
if let Ok(resource) = resource {
|
||||
let expire = resource.expire;
|
||||
if expire > chrono::Utc::now().timestamp() {
|
||||
let result = resource.value;
|
||||
let logs =
|
||||
"Job skipped because args & path found in cache and not expired".to_string();
|
||||
let cached_resource_value_maybe = get_cached_resource_value_if_valid(
|
||||
db,
|
||||
&authed_client,
|
||||
&job.id,
|
||||
&job.workspace_id,
|
||||
&cached_res_path,
|
||||
)
|
||||
.await;
|
||||
if let Some(cached_resource_value) = cached_resource_value_maybe {
|
||||
let logs = "Job skipped because args & path found in cache and not expired".to_string();
|
||||
job_completed_tx
|
||||
.send(JobCompleted {
|
||||
job: job,
|
||||
result: cached_resource_value,
|
||||
logs,
|
||||
mem_peak: 0,
|
||||
canceled_by: None,
|
||||
success: true,
|
||||
cached_res_path: None,
|
||||
token: authed_client.token,
|
||||
})
|
||||
.await
|
||||
.expect("send job completed");
|
||||
|
||||
job_completed_tx
|
||||
.send(JobCompleted {
|
||||
job: job,
|
||||
result,
|
||||
logs,
|
||||
mem_peak: 0,
|
||||
canceled_by: None,
|
||||
success: true,
|
||||
cached_res_path: None,
|
||||
token: authed_client.token,
|
||||
})
|
||||
.await
|
||||
.expect("send job completed");
|
||||
|
||||
return Ok(());
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
match job.job_kind {
|
||||
|
||||
@@ -633,7 +633,8 @@ pub async fn update_flow_status_after_job_completion_internal<
|
||||
} else {
|
||||
if flow_job.cache_ttl.is_some() {
|
||||
let cached_res_path = {
|
||||
let args_hash = hash_args(&flow_job.args);
|
||||
let args_hash =
|
||||
hash_args(db, client, w_id, job_id_for_status, &flow_job.args).await;
|
||||
let flow_path = flow_job.script_path();
|
||||
let version_hash = if let Some(rc) = flow_job.raw_flow.as_ref() {
|
||||
use std::hash::Hasher;
|
||||
@@ -648,7 +649,7 @@ pub async fn update_flow_status_after_job_completion_internal<
|
||||
format!("{flow_path}/cache/{version_hash}/{args_hash}")
|
||||
};
|
||||
|
||||
save_in_cache(db, &flow_job, cached_res_path, &nresult).await;
|
||||
save_in_cache(db, client, &flow_job, cached_res_path, &nresult).await;
|
||||
}
|
||||
let success = success && !is_failure_step && !skip_error_handler;
|
||||
if success {
|
||||
|
||||
Reference in New Issue
Block a user