From 0003673e8d7765179753dd47ef2cd77c7a438dc0 Mon Sep 17 00:00:00 2001 From: Guillaume Bouvignies Date: Thu, 1 Feb 2024 16:03:30 +0100 Subject: [PATCH] feat: migrate s3 client to object_store (#3116) * feat: migrate s3 client to object_store * remove multipart from open API * Remove multipart in favor of a single stream * progress report * add progress reader on server side * small nit fix * fix read chunk * Fix TS and python SDK * Fix download button * Fix download button object viewer * fix list * Better errors * export loadS3FileContent * revert changes SDK * fix browser * small file list unavailable fix * Old endpoints throws informative error messages * Typescript SDK uses raw fetch * update python SDK * Error if uploaded file > 50Mb * revert python SDL changes * Update python SDK method docs --------- Co-authored-by: Ruben Fiszel Co-authored-by: Ruben Fiszel --- backend/Cargo.lock | 581 +----------- backend/Cargo.toml | 5 +- backend/windmill-api/Cargo.toml | 2 +- backend/windmill-api/openapi-deref.yaml | 613 ++++++------- backend/windmill-api/openapi.yaml | 153 +--- backend/windmill-api/src/job_helpers.rs | 833 +++++++----------- backend/windmill-common/Cargo.toml | 3 +- backend/windmill-common/src/s3_helpers.rs | 79 +- .../src/lib/components/S3FilePicker.svelte | 196 +++-- .../common/fileDownload/FileDownload.svelte | 30 +- .../common/fileUpload/FileUpload.svelte | 211 +++-- .../propertyPicker/ObjectViewer.svelte | 51 +- .../(logged)/workspace_settings/+page.svelte | 92 +- python-client/tests/wmill_client_test.py | 21 + python-client/wmill/wmill/client.py | 78 +- python-client/wmill/wmill/s3_reader.py | 157 +--- typescript-client/build.sh | 2 +- typescript-client/client.ts | 141 +-- 18 files changed, 1162 insertions(+), 2086 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index e00a522ca3..985956c0bb 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -399,7 +399,7 @@ checksum = "beec3f8fb8f710b7be84ccd1716e17f38f2868168355cab5f2f168ae988e767e" dependencies = [ "base64 0.21.7", "bytes", - "http 0.2.11", + "http", "rand 0.8.5", "reqwest", "serde", @@ -576,368 +576,6 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" -[[package]] -name = "aws-config" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b30c39ebe61f75d1b3785362b1586b41991873c9ab3e317a9181c246fb71d82" -dependencies = [ - "aws-credential-types", - "aws-runtime", - "aws-sdk-sso", - "aws-sdk-ssooidc", - "aws-sdk-sts", - "aws-smithy-async", - "aws-smithy-http", - "aws-smithy-json", - "aws-smithy-runtime", - "aws-smithy-runtime-api", - "aws-smithy-types", - "aws-types", - "bytes", - "fastrand 2.0.1", - "hex", - "http 0.2.11", - "hyper", - "ring 0.17.7", - "time", - "tokio", - "tracing", - "zeroize", -] - -[[package]] -name = "aws-credential-types" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33cc49dcdd31c8b6e79850a179af4c367669150c7ac0135f176c61bec81a70f7" -dependencies = [ - "aws-smithy-async", - "aws-smithy-runtime-api", - "aws-smithy-types", - "zeroize", -] - -[[package]] -name = "aws-runtime" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb031bff99877c26c28895766f7bb8484a05e24547e370768d6cc9db514662aa" -dependencies = [ - "aws-credential-types", - "aws-sigv4", - "aws-smithy-async", - "aws-smithy-eventstream", - "aws-smithy-http", - "aws-smithy-runtime-api", - "aws-smithy-types", - "aws-types", - "bytes", - "fastrand 2.0.1", - "http 0.2.11", - "http-body", - "percent-encoding", - "pin-project-lite", - "tracing", - "uuid 1.7.0", -] - -[[package]] -name = "aws-sdk-s3" -version = "1.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "951f7730f51a2155c711c85c79f337fbc02a577fa99d2a0a8059acfce5392113" -dependencies = [ - "aws-credential-types", - "aws-runtime", - "aws-sigv4", - "aws-smithy-async", - "aws-smithy-checksums", - "aws-smithy-eventstream", - "aws-smithy-http", - "aws-smithy-json", - "aws-smithy-runtime", - "aws-smithy-runtime-api", - "aws-smithy-types", - "aws-smithy-xml", - "aws-types", - "bytes", - "http 0.2.11", - "http-body", - "once_cell", - "percent-encoding", - "regex-lite", - "tracing", - "url", -] - -[[package]] -name = "aws-sdk-sso" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f486420a66caad72635bc2ce0ff6581646e0d32df02aa39dc983bfe794955a5b" -dependencies = [ - "aws-credential-types", - "aws-runtime", - "aws-smithy-async", - "aws-smithy-http", - "aws-smithy-json", - "aws-smithy-runtime", - "aws-smithy-runtime-api", - "aws-smithy-types", - "aws-types", - "bytes", - "http 0.2.11", - "once_cell", - "regex-lite", - "tracing", -] - -[[package]] -name = "aws-sdk-ssooidc" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39ddccf01d82fce9b4a15c8ae8608211ee7db8ed13a70b514bbfe41df3d24841" -dependencies = [ - "aws-credential-types", - "aws-runtime", - "aws-smithy-async", - "aws-smithy-http", - "aws-smithy-json", - "aws-smithy-runtime", - "aws-smithy-runtime-api", - "aws-smithy-types", - "aws-types", - "bytes", - "http 0.2.11", - "once_cell", - "regex-lite", - "tracing", -] - -[[package]] -name = "aws-sdk-sts" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a591f8c7e6a621a501b2b5d2e88e1697fcb6274264523a6ad4d5959889a41ce" -dependencies = [ - "aws-credential-types", - "aws-runtime", - "aws-smithy-async", - "aws-smithy-http", - "aws-smithy-json", - "aws-smithy-query", - "aws-smithy-runtime", - "aws-smithy-runtime-api", - "aws-smithy-types", - "aws-smithy-xml", - "aws-types", - "http 0.2.11", - "once_cell", - "regex-lite", - "tracing", -] - -[[package]] -name = "aws-sigv4" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c371c6b0ac54d4605eb6f016624fb5c7c2925d315fdf600ac1bf21b19d5f1742" -dependencies = [ - "aws-credential-types", - "aws-smithy-eventstream", - "aws-smithy-http", - "aws-smithy-runtime-api", - "aws-smithy-types", - "bytes", - "crypto-bigint 0.5.5", - "form_urlencoded", - "hex", - "hmac", - "http 0.2.11", - "http 1.0.0", - "once_cell", - "p256 0.11.1", - "percent-encoding", - "ring 0.17.7", - "sha2 0.10.8", - "subtle", - "time", - "tracing", - "zeroize", -] - -[[package]] -name = "aws-smithy-async" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72ee2d09cce0ef3ae526679b522835d63e75fb427aca5413cd371e490d52dcc6" -dependencies = [ - "futures-util", - "pin-project-lite", - "tokio", -] - -[[package]] -name = "aws-smithy-checksums" -version = "0.60.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "be2acd1b9c6ae5859999250ed5a62423aedc5cf69045b844432de15fa2f31f2b" -dependencies = [ - "aws-smithy-http", - "aws-smithy-types", - "bytes", - "crc32c", - "crc32fast", - "hex", - "http 0.2.11", - "http-body", - "md-5 0.10.6", - "pin-project-lite", - "sha1", - "sha2 0.10.8", - "tracing", -] - -[[package]] -name = "aws-smithy-eventstream" -version = "0.60.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6363078f927f612b970edf9d1903ef5cef9a64d1e8423525ebb1f0a1633c858" -dependencies = [ - "aws-smithy-types", - "bytes", - "crc32fast", -] - -[[package]] -name = "aws-smithy-http" -version = "0.60.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dab56aea3cd9e1101a0a999447fb346afb680ab1406cebc44b32346e25b4117d" -dependencies = [ - "aws-smithy-eventstream", - "aws-smithy-runtime-api", - "aws-smithy-types", - "bytes", - "bytes-utils", - "futures-core", - "http 0.2.11", - "http-body", - "once_cell", - "percent-encoding", - "pin-project-lite", - "pin-utils", - "tracing", -] - -[[package]] -name = "aws-smithy-json" -version = "0.60.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd3898ca6518f9215f62678870064398f00031912390efd03f1f6ef56d83aa8e" -dependencies = [ - "aws-smithy-types", -] - -[[package]] -name = "aws-smithy-query" -version = "0.60.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bda4b1dfc9810e35fba8a620e900522cd1bd4f9578c446e82f49d1ce41d2e9f9" -dependencies = [ - "aws-smithy-types", - "urlencoding", -] - -[[package]] -name = "aws-smithy-runtime" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fafdab38f40ad7816e7da5dec279400dd505160780083759f01441af1bbb10ea" -dependencies = [ - "aws-smithy-async", - "aws-smithy-http", - "aws-smithy-runtime-api", - "aws-smithy-types", - "bytes", - "fastrand 2.0.1", - "h2", - "http 0.2.11", - "http-body", - "hyper", - "hyper-rustls", - "once_cell", - "pin-project-lite", - "pin-utils", - "rustls 0.21.10", - "tokio", - "tracing", -] - -[[package]] -name = "aws-smithy-runtime-api" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c18276dd28852f34b3bf501f4f3719781f4999a51c7bff1a5c6dc8c4529adc29" -dependencies = [ - "aws-smithy-async", - "aws-smithy-types", - "bytes", - "http 0.2.11", - "pin-project-lite", - "tokio", - "tracing", - "zeroize", -] - -[[package]] -name = "aws-smithy-types" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb3e134004170d3303718baa2a4eb4ca64ee0a1c0a7041dca31b38be0fb414f3" -dependencies = [ - "base64-simd", - "bytes", - "bytes-utils", - "futures-core", - "http 0.2.11", - "http-body", - "itoa", - "num-integer", - "pin-project-lite", - "pin-utils", - "ryu", - "serde", - "time", - "tokio", - "tokio-util", -] - -[[package]] -name = "aws-smithy-xml" -version = "0.60.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8604a11b25e9ecaf32f9aa56b9fe253c5e2f606a3477f0071e96d3155a5ed218" -dependencies = [ - "xmlparser", -] - -[[package]] -name = "aws-types" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "789bbe008e65636fe1b6dbbb374c40c8960d1232b96af5ff4aec349f9c4accf4" -dependencies = [ - "aws-credential-types", - "aws-smithy-async", - "aws-smithy-runtime-api", - "aws-smithy-types", - "http 0.2.11", - "rustc_version 0.4.0", - "tracing", -] - [[package]] name = "axum" version = "0.6.20" @@ -950,7 +588,7 @@ dependencies = [ "bytes", "futures-util", "headers", - "http 0.2.11", + "http", "http-body", "hyper", "itoa", @@ -980,7 +618,7 @@ dependencies = [ "async-trait", "bytes", "futures-util", - "http 0.2.11", + "http", "http-body", "mime", "rustversion", @@ -1003,12 +641,6 @@ dependencies = [ "rustc-demangle", ] -[[package]] -name = "base16ct" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "349a06037c7bf932dd7e7d1f653678b2038b9ad46a74102f1fc7bd7872678cce" - [[package]] name = "base16ct" version = "0.2.0" @@ -1332,16 +964,6 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2bd12c1caf447e69cd4528f47f94d203fd2582878ecb9e9465484c4148a8223" -[[package]] -name = "bytes-utils" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35" -dependencies = [ - "bytes", - "either", -] - [[package]] name = "bzip2" version = "0.4.4" @@ -1755,15 +1377,6 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" -[[package]] -name = "crc32c" -version = "0.6.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8f48d60e5b4d2c53d5c2b1d8a58c849a70ae5e5509b08a48d047e3b65714a74" -dependencies = [ - "rustc_version 0.4.0", -] - [[package]] name = "crc32fast" version = "1.3.2" @@ -1868,18 +1481,6 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7a81dae078cea95a014a339291cec439d2f232ebe854a9d672b796c6afafa9b7" -[[package]] -name = "crypto-bigint" -version = "0.4.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef2b4b23cddf68b89b8f8069890e8c270d54e2d5fe1b143820234805e4cb17ef" -dependencies = [ - "generic-array", - "rand_core 0.6.4", - "subtle", - "zeroize", -] - [[package]] name = "crypto-bigint" version = "0.5.5" @@ -2152,7 +1753,7 @@ dependencies = [ "deno_core", "deno_tls", "dyn-clone", - "http 0.2.11", + "http", "reqwest", "serde", "tokio", @@ -2450,18 +2051,6 @@ dependencies = [ "reborrow", ] -[[package]] -name = "ecdsa" -version = "0.14.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413301934810f597c1d19ca71c8710e99a3f1ba28a0d2ebc01551a2daeea3c5c" -dependencies = [ - "der 0.6.1", - "elliptic-curve 0.12.3", - "rfc6979 0.3.1", - "signature 1.6.4", -] - [[package]] name = "ecdsa" version = "0.16.9" @@ -2470,8 +2059,8 @@ checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca" dependencies = [ "der 0.7.8", "digest 0.10.7", - "elliptic-curve 0.13.8", - "rfc6979 0.4.0", + "elliptic-curve", + "rfc6979", "signature 2.2.0", "spki 0.7.3", ] @@ -2509,43 +2098,23 @@ dependencies = [ "serde", ] -[[package]] -name = "elliptic-curve" -version = "0.12.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7bb888ab5300a19b8e5bceef25ac745ad065f3c9f7efc6de1b91958110891d3" -dependencies = [ - "base16ct 0.1.1", - "crypto-bigint 0.4.9", - "der 0.6.1", - "digest 0.10.7", - "ff 0.12.1", - "generic-array", - "group 0.12.1", - "pkcs8 0.9.0", - "rand_core 0.6.4", - "sec1 0.3.0", - "subtle", - "zeroize", -] - [[package]] name = "elliptic-curve" version = "0.13.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47" dependencies = [ - "base16ct 0.2.0", - "crypto-bigint 0.5.5", + "base16ct", + "crypto-bigint", "digest 0.10.7", - "ff 0.13.0", + "ff", "generic-array", - "group 0.13.0", + "group", "hkdf", "pem-rfc7468 0.7.0", "pkcs8 0.10.2", "rand_core 0.6.4", - "sec1 0.7.3", + "sec1", "subtle", "zeroize", ] @@ -2787,16 +2356,6 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25cbce373ec4653f1a01a31e8a5e5ec0c622dc27ff9c4e6606eefef5cbbed4a5" -[[package]] -name = "ff" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d013fc25338cc558c5c2cfbad646908fb23591e2404481826742b651c9af7160" -dependencies = [ - "rand_core 0.6.4", - "subtle", -] - [[package]] name = "ff" version = "0.13.0" @@ -3364,24 +2923,13 @@ dependencies = [ "unic-ucd-category", ] -[[package]] -name = "group" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5dfbfb3a6cfbd390d5c9564ab283a0349b9b9fcd46a706c1eb10e0db70bfbac7" -dependencies = [ - "ff 0.12.1", - "rand_core 0.6.4", - "subtle", -] - [[package]] name = "group" version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0f9ef7462f7c099f518d754361858f86d8a07af53ba9af0fe635bbccb151a63" dependencies = [ - "ff 0.13.0", + "ff", "rand_core 0.6.4", "subtle", ] @@ -3397,7 +2945,7 @@ dependencies = [ "futures-core", "futures-sink", "futures-util", - "http 0.2.11", + "http", "indexmap 2.2.1", "slab", "tokio", @@ -3466,7 +3014,7 @@ dependencies = [ "base64 0.21.7", "bytes", "headers-core", - "http 0.2.11", + "http", "httpdate", "mime", "sha1", @@ -3478,7 +3026,7 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7f66481bfee273957b1f20485a4ff3362987f85b2c236580d81b4eb7a326429" dependencies = [ - "http 0.2.11", + "http", ] [[package]] @@ -3557,17 +3105,6 @@ dependencies = [ "itoa", ] -[[package]] -name = "http" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b32afd38673a8016f7c9ae69e5af41a58f81b1d31689040f2f1959594ce194ea" -dependencies = [ - "bytes", - "fnv", - "itoa", -] - [[package]] name = "http-body" version = "0.4.6" @@ -3575,7 +3112,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" dependencies = [ "bytes", - "http 0.2.11", + "http", "pin-project-lite", ] @@ -3595,7 +3132,7 @@ dependencies = [ "async-channel 1.9.0", "base64 0.13.1", "futures-lite 1.13.0", - "http 0.2.11", + "http", "infer", "pin-project-lite", "rand 0.7.3", @@ -3635,7 +3172,7 @@ dependencies = [ "futures-core", "futures-util", "h2", - "http 0.2.11", + "http", "http-body", "httparse", "httpdate", @@ -3655,9 +3192,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec3efd23720e2049821a693cbc7e65ea87c72f1c58ff2f9522ff332b1491e590" dependencies = [ "futures-util", - "http 0.2.11", + "http", "hyper", - "log", "rustls 0.21.10", "rustls-native-certs", "tokio", @@ -4690,7 +4226,7 @@ dependencies = [ "base64 0.13.1", "chrono", "getrandom 0.2.12", - "http 0.2.11", + "http", "rand 0.8.5", "reqwest", "serde", @@ -4795,11 +4331,11 @@ dependencies = [ "dyn-clone", "ed25519-dalek", "hmac", - "http 0.2.11", + "http", "itertools 0.10.5", "log", "oauth2", - "p256 0.13.2", + "p256", "p384", "rand 0.8.5", "rsa 0.9.6", @@ -4887,25 +4423,14 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" -[[package]] -name = "p256" -version = "0.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51f44edd08f51e2ade572f141051021c5af22677e42b7dd28a88155151c33594" -dependencies = [ - "ecdsa 0.14.8", - "elliptic-curve 0.12.3", - "sha2 0.10.8", -] - [[package]] name = "p256" version = "0.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c9863ad85fa8f4460f9c48cb909d38a0d689dba1f6f6988a5e3e0d31071bcd4b" dependencies = [ - "ecdsa 0.16.9", - "elliptic-curve 0.13.8", + "ecdsa", + "elliptic-curve", "primeorder", "sha2 0.10.8", ] @@ -4916,8 +4441,8 @@ version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "70786f51bcc69f6a4c0360e063a4cac5419ef7c5cd5b3c99ad70f3be5ba79209" dependencies = [ - "ecdsa 0.16.9", - "elliptic-curve 0.13.8", + "ecdsa", + "elliptic-curve", "primeorder", "sha2 0.10.8", ] @@ -5780,7 +5305,7 @@ version = "0.13.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "353e1ca18966c16d9deb1c69278edbc5f194139612772bd9537af60ac231e1e6" dependencies = [ - "elliptic-curve 0.13.8", + "elliptic-curve", ] [[package]] @@ -5881,7 +5406,7 @@ source = "git+https://github.com/oxidecomputer/progenitor?rev=3d96016ae8d422e905 dependencies = [ "getopts", "heck", - "http 0.2.11", + "http", "indexmap 1.9.3", "openapiv3", "proc-macro2", @@ -6269,12 +5794,6 @@ dependencies = [ "regex-syntax 0.8.2", ] -[[package]] -name = "regex-lite" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30b661b2f27137bdbc16f00eda72866a92bb28af1753ffbd56744fb6e2e9cd8e" - [[package]] name = "regex-syntax" version = "0.6.29" @@ -6325,7 +5844,7 @@ dependencies = [ "futures-core", "futures-util", "h2", - "http 0.2.11", + "http", "http-body", "hyper", "hyper-rustls", @@ -6360,17 +5879,6 @@ dependencies = [ "winreg", ] -[[package]] -name = "rfc6979" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7743f17af12fa0b03b803ba12cd6a8d9483a587e89c69445e3909655c0b9fabb" -dependencies = [ - "crypto-bigint 0.4.9", - "hmac", - "zeroize", -] - [[package]] name = "rfc6979" version = "0.4.0" @@ -6892,27 +6400,13 @@ version = "4.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b" -[[package]] -name = "sec1" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3be24c1842290c45df0a7bf069e0c268a747ad05a192f2fd7dcfdbc1cba40928" -dependencies = [ - "base16ct 0.1.1", - "der 0.6.1", - "generic-array", - "pkcs8 0.9.0", - "subtle", - "zeroize", -] - [[package]] name = "sec1" version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc" dependencies = [ - "base16ct 0.2.0", + "base16ct", "der 0.7.8", "generic-array", "pkcs8 0.10.2", @@ -8791,7 +8285,7 @@ dependencies = [ "axum-core", "cookie", "futures-util", - "http 0.2.11", + "http", "parking_lot", "pin-project-lite", "tower-layer", @@ -8808,7 +8302,7 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http 0.2.11", + "http", "http-body", "http-range-header", "pin-project-lite", @@ -9617,7 +9111,6 @@ dependencies = [ "async-recursion", "async-stripe", "async_zip", - "aws-sdk-s3", "axum", "base64 0.21.7", "bytes", @@ -9641,6 +9134,7 @@ dependencies = [ "mime_guess", "object_store", "openidconnect", + "pin-project", "polars", "polars-io", "prometheus", @@ -9717,8 +9211,6 @@ name = "windmill-common" version = "1.258.4" dependencies = [ "anyhow", - "aws-config", - "aws-sdk-s3", "axum", "chrono", "cron", @@ -9729,6 +9221,7 @@ dependencies = [ "itertools 0.12.1", "lazy_static", "magic-crypt", + "object_store", "prometheus", "rand 0.8.5", "regex", @@ -10155,12 +9648,6 @@ dependencies = [ "rustix 0.38.30", ] -[[package]] -name = "xmlparser" -version = "0.13.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" - [[package]] name = "xxhash-rust" version = "0.8.8" diff --git a/backend/Cargo.toml b/backend/Cargo.toml index ff1ddc2094..f195623846 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -211,10 +211,9 @@ candle-core = "0.3.0" candle-transformers = "0.3.0" candle-nn = "0.3.0" tiberius = { version = "0.12.2", default-features = false, features = ["rustls", "tds73", "chrono", "sql-browser-tokio"] } -aws-sdk-s3 = "1.5.0" -aws-config = "1.0.3" +pin-project = "1" polars = { version = "0.35.4", features = ["lazy", "parquet", "aws", "csv", "dtype-full", "serde", "strings", "extract_groups"] } polars-io = { version = "0.35.4", features = ["csv"] } -object_store = { version = "0.8.0", features = ["aws"] } +object_store = { version = "0.8.0", features = ["aws", "azure"] } openidconnect = { version = "3.4.0" } diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 0bd21d2a03..61f5c7d805 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -86,8 +86,8 @@ tokenizers.workspace = true candle-core.workspace = true candle-transformers.workspace = true candle-nn.workspace = true -aws-sdk-s3.workspace = true polars.workspace = true polars-io.workspace = true object_store.workspace = true openidconnect = { workspace = true, optional = true} +pin-project.workspace = true diff --git a/backend/windmill-api/openapi-deref.yaml b/backend/windmill-api/openapi-deref.yaml index 71cb1d07e5..6373cf032b 100644 --- a/backend/windmill-api/openapi-deref.yaml +++ b/backend/windmill-api/openapi-deref.yaml @@ -1,6 +1,6 @@ openapi: 3.0.3 info: - version: 1.253.4 + version: 1.258.4 title: Windmill API contact: name: Windmill Team @@ -231,34 +231,34 @@ paths: - name: before description: filter on created before (exclusive) timestamp in: query - schema: &ref_105 + schema: &ref_103 type: string format: date-time - name: after description: filter on created after (exclusive) timestamp in: query - schema: &ref_104 + schema: &ref_102 type: string format: date-time - name: username description: filter on exact username of user in: query - schema: &ref_106 + schema: &ref_104 type: string - name: operation description: filter on exact or prefix name of operation in: query - schema: &ref_107 + schema: &ref_105 type: string - name: resource description: filter on exact or prefix name of resource in: query - schema: &ref_108 + schema: &ref_106 type: string - name: action_kind description: filter on type of operation in: query - schema: &ref_109 + schema: &ref_107 type: string enum: - Create @@ -290,12 +290,12 @@ paths: application/json: schema: type: object - properties: &ref_122 + properties: &ref_120 email: type: string password: type: string - required: &ref_123 + required: &ref_121 - email - password responses: @@ -351,14 +351,14 @@ paths: application/json: schema: type: object - properties: &ref_124 + properties: &ref_122 email: type: string username: type: string is_admin: type: boolean - required: &ref_125 + required: &ref_123 - email - username - is_admin @@ -393,7 +393,7 @@ paths: application/json: schema: type: object - properties: &ref_126 + properties: &ref_124 is_admin: type: boolean operator: @@ -616,7 +616,7 @@ paths: application/json: schema: type: object - properties: &ref_154 + properties: &ref_152 email: type: string workspaces: @@ -634,7 +634,7 @@ paths: - id - name - username - required: &ref_155 + required: &ref_153 - email - workspaces /workspaces/list_as_superadmin: @@ -676,14 +676,14 @@ paths: application/json: schema: type: object - properties: &ref_156 + properties: &ref_154 id: type: string name: type: string username: type: string - required: &ref_157 + required: &ref_155 - id - name - username @@ -1124,7 +1124,7 @@ paths: type: string usage: type: object - properties: &ref_121 + properties: &ref_119 executions: type: number required: &ref_9 @@ -1496,6 +1496,8 @@ paths: type: string auto_invite_operator: type: boolean + auto_add: + type: boolean plan: type: string customer_id: @@ -1723,6 +1725,8 @@ paths: type: boolean invite_all: type: boolean + auto_add: + type: boolean responses: '200': description: status @@ -2031,7 +2035,7 @@ paths: application/json: schema: type: object - properties: &ref_129 + properties: &ref_127 label: type: string expiration: @@ -2061,7 +2065,7 @@ paths: application/json: schema: type: object - properties: &ref_130 + properties: &ref_128 label: type: string expiration: @@ -2069,7 +2073,7 @@ paths: format: date-time impersonate_email: type: string - required: &ref_131 + required: &ref_129 - impersonate_email responses: '201': @@ -2117,7 +2121,7 @@ paths: type: array items: type: object - properties: &ref_127 + properties: &ref_125 label: type: string expiration: @@ -2135,7 +2139,7 @@ paths: type: array items: type: string - required: &ref_128 + required: &ref_126 - token_prefix - created_at - last_used_at @@ -2184,7 +2188,7 @@ paths: application/json: schema: type: object - properties: &ref_134 + properties: &ref_132 path: type: string value: @@ -2197,7 +2201,7 @@ paths: type: integer is_oauth: type: boolean - required: &ref_135 + required: &ref_133 - path - value - is_secret @@ -2282,7 +2286,7 @@ paths: application/json: schema: type: object - properties: &ref_136 + properties: &ref_134 path: type: string value: @@ -2452,14 +2456,14 @@ paths: type: array items: type: object - properties: &ref_132 + properties: &ref_130 name: type: string value: type: string description: type: string - required: &ref_133 + required: &ref_131 - name - value - description @@ -2570,7 +2574,7 @@ paths: application/json: schema: type: object - properties: &ref_170 + properties: &ref_168 access_token: type: string expires_in: @@ -2581,7 +2585,7 @@ paths: type: array items: type: string - required: &ref_171 + required: &ref_169 - access_token /w/{workspace}/oauth/create_account: post: @@ -2759,7 +2763,7 @@ paths: application/json: schema: type: object - properties: &ref_139 + properties: &ref_137 path: type: string value: {} @@ -2767,7 +2771,7 @@ paths: type: string resource_type: type: string - required: &ref_140 + required: &ref_138 - path - value - resource_type @@ -2822,7 +2826,7 @@ paths: application/json: schema: type: object - properties: &ref_141 + properties: &ref_139 path: type: string description: @@ -2888,7 +2892,7 @@ paths: application/json: schema: type: object - properties: &ref_142 + properties: &ref_140 workspace_id: type: string path: @@ -2904,7 +2908,7 @@ paths: type: object additionalProperties: type: boolean - required: &ref_143 + required: &ref_141 - path - resource_type - is_oauth @@ -3016,7 +3020,7 @@ paths: type: array items: type: object - properties: &ref_144 + properties: &ref_142 workspace_id: type: string path: @@ -3042,7 +3046,7 @@ paths: type: boolean account: type: number - required: &ref_145 + required: &ref_143 - path - resource_type - is_oauth @@ -3187,7 +3191,7 @@ paths: application/json: schema: type: object - properties: &ref_146 + properties: &ref_144 schema: {} description: type: string @@ -3431,32 +3435,32 @@ paths: id: type: string value: - oneOf: &ref_209 + oneOf: &ref_207 - type: object - properties: &ref_193 + properties: &ref_191 input_transforms: type: object additionalProperties: oneOf: &ref_26 - type: object - properties: &ref_189 + properties: &ref_187 value: {} type: type: string enum: - javascript - required: &ref_190 + required: &ref_188 - expr - type - type: object - properties: &ref_191 + properties: &ref_189 expr: type: string type: type: string enum: - javascript - required: &ref_192 + required: &ref_190 - expr - type discriminator: &ref_27 @@ -3496,13 +3500,13 @@ paths: type: number concurrency_time_window_s: type: number - required: &ref_194 + required: &ref_192 - type - content - language - input_transforms - type: object - properties: &ref_195 + properties: &ref_193 input_transforms: type: object additionalProperties: @@ -3516,12 +3520,12 @@ paths: type: string enum: - script - required: &ref_196 + required: &ref_194 - type - path - input_transforms - type: object - properties: &ref_197 + properties: &ref_195 input_transforms: type: object additionalProperties: @@ -3533,12 +3537,12 @@ paths: type: string enum: - flow - required: &ref_198 + required: &ref_196 - type - path - input_transforms - type: object - properties: &ref_199 + properties: &ref_197 modules: type: array items: @@ -3560,13 +3564,13 @@ paths: type: boolean parallelism: type: integer - required: &ref_200 + required: &ref_198 - modules - iterator - skip_failures - type - type: object - properties: &ref_201 + properties: &ref_199 branches: type: array items: @@ -3597,12 +3601,12 @@ paths: type: string enum: - branchone - required: &ref_202 + required: &ref_200 - branches - default - type - type: object - properties: &ref_203 + properties: &ref_201 branches: type: array items: @@ -3627,28 +3631,28 @@ paths: - branchall parallel: type: boolean - required: &ref_204 + required: &ref_202 - branches - type - type: object - properties: &ref_205 + properties: &ref_203 type: type: string enum: - identity flow: type: boolean - required: &ref_206 + required: &ref_204 - type - type: object - properties: &ref_207 + properties: &ref_205 type: type: string enum: - graphql - required: &ref_208 + required: &ref_206 - type - discriminator: &ref_210 + discriminator: &ref_208 propertyName: type mapping: rawscript: '#/components/schemas/RawScript' @@ -4643,7 +4647,7 @@ paths: content: application/json: schema: - allOf: &ref_113 + allOf: &ref_111 - type: object properties: *ref_35 required: *ref_36 @@ -4681,12 +4685,12 @@ paths: type: array items: type: object - properties: &ref_114 + properties: &ref_112 script_hash: type: string deployment_msg: type: string - required: &ref_115 + required: &ref_113 - script_hash /w/{workspace}/scripts/history_update/h/{hash}/p/{path}: post: @@ -4762,7 +4766,7 @@ paths: - name: token in: path required: true - schema: &ref_99 + schema: &ref_97 type: string - name: path in: path @@ -5373,7 +5377,7 @@ paths: properties: *ref_44 required: *ref_45 - type: object - properties: &ref_158 + properties: &ref_156 workspace_id: type: string path: @@ -5403,7 +5407,7 @@ paths: type: boolean timeout: type: number - required: &ref_159 + required: &ref_157 - path - edited_by - edited_at @@ -5765,7 +5769,7 @@ paths: type: array items: type: object - properties: &ref_165 + properties: &ref_163 workspace_id: type: string path: @@ -5783,7 +5787,7 @@ paths: edited_at: type: string format: date-time - required: &ref_166 + required: &ref_164 - workspace_id - path - summary @@ -5826,7 +5830,7 @@ paths: - name: version in: path required: true - schema: &ref_98 + schema: &ref_96 type: number - name: path in: path @@ -5919,7 +5923,7 @@ paths: type: array items: type: object - properties: &ref_163 + properties: &ref_161 id: type: integer workspace_id: @@ -5945,7 +5949,7 @@ paths: - viewer - publisher - anonymous - required: &ref_164 + required: &ref_162 - id - workspace_id - path @@ -6120,7 +6124,7 @@ paths: content: application/json: schema: - allOf: &ref_167 + allOf: &ref_165 - type: object properties: *ref_49 required: *ref_50 @@ -6153,12 +6157,12 @@ paths: type: array items: type: object - properties: &ref_168 + properties: &ref_166 version: type: integer deployment_msg: type: string - required: &ref_169 + required: &ref_167 - version /w/{workspace}/apps/history_update/a/{id}/v/{version}: post: @@ -6178,7 +6182,7 @@ paths: - name: version in: path required: true - schema: &ref_100 + schema: &ref_98 type: integer requestBody: description: App deployment message @@ -6775,7 +6779,7 @@ paths: application/json: schema: type: object - properties: &ref_137 + properties: &ref_135 content: type: string path: @@ -6809,7 +6813,9 @@ paths: - http dedicated_worker: type: boolean - required: &ref_138 + lock: + type: string + required: &ref_136 - args responses: '201': @@ -6842,7 +6848,7 @@ paths: type: array items: type: object - properties: &ref_185 + properties: &ref_183 raw_code: type: string path: @@ -6863,7 +6869,7 @@ paths: - graphql - nativets - bun - required: &ref_186 + required: &ref_184 - raw_code - path - language @@ -6924,7 +6930,7 @@ paths: application/json: schema: type: object - properties: &ref_160 + properties: &ref_158 value: type: object properties: *ref_51 @@ -6938,7 +6944,7 @@ paths: type: string restarted_from: type: object - properties: &ref_162 + properties: &ref_160 flow_job_id: type: string format: uuid @@ -6946,7 +6952,7 @@ paths: type: string branch_or_iteration_n: type: integer - required: &ref_161 + required: &ref_159 - value - content - args @@ -7035,7 +7041,7 @@ paths: - name: suspended description: filter on suspended jobs in: query - schema: &ref_103 + schema: &ref_101 type: boolean - name: running description: filter on running jobs @@ -7602,7 +7608,7 @@ paths: filter on created_at for non non started job and started_at otherwise before (inclusive) timestamp in: query - schema: &ref_102 + schema: &ref_100 type: string format: date-time - name: running @@ -7618,7 +7624,7 @@ paths: filter on created_at for non non started job and started_at otherwise after (exclusive) timestamp in: query - schema: &ref_101 + schema: &ref_99 type: string format: date-time - name: job_kinds @@ -7842,7 +7848,7 @@ paths: schema: *ref_74 - name: get_started in: query - schema: &ref_111 + schema: &ref_109 type: boolean responses: '200': @@ -8349,7 +8355,7 @@ paths: application/json: schema: type: object - properties: &ref_148 + properties: &ref_146 path: type: string schedule: @@ -8392,7 +8398,7 @@ paths: type: string tag: type: string - required: &ref_149 + required: &ref_147 - path - schedule - timezone @@ -8428,7 +8434,7 @@ paths: application/json: schema: type: object - properties: &ref_150 + properties: &ref_148 schedule: type: string timezone: @@ -8463,7 +8469,7 @@ paths: type: string tag: type: string - required: &ref_151 + required: &ref_149 - schedule - timezone - script_path @@ -8709,7 +8715,7 @@ paths: schema: type: array items: - allOf: &ref_147 + allOf: &ref_145 - type: object properties: *ref_79 required: *ref_80 @@ -9513,7 +9519,7 @@ paths: type: array items: type: object - properties: &ref_152 + properties: &ref_150 worker: type: string worker_instance: @@ -9535,7 +9541,7 @@ paths: type: string wm_version: type: string - required: &ref_153 + required: &ref_151 - worker - worker_instance - ping_at @@ -9917,7 +9923,7 @@ paths: in: query schema: &ref_91 type: string - enum: &ref_120 + enum: &ref_118 - ScriptHash - ScriptPath - FlowPath @@ -10000,12 +10006,12 @@ paths: application/json: schema: type: object - properties: &ref_116 + properties: &ref_114 name: type: string args: type: object - required: &ref_117 + required: &ref_115 - name - args - created_by @@ -10035,14 +10041,14 @@ paths: application/json: schema: type: object - properties: &ref_118 + properties: &ref_116 id: type: string name: type: string is_public: type: boolean - required: &ref_119 + required: &ref_117 - id - name - is_public @@ -10068,7 +10074,7 @@ paths: - name: input in: path required: true - schema: &ref_110 + schema: &ref_108 type: string responses: '200': @@ -10384,10 +10390,10 @@ paths: type: array items: type: object - properties: &ref_172 + properties: &ref_170 s3: type: string - required: &ref_173 + required: &ref_171 - s3 restricted_access: type: boolean @@ -10416,7 +10422,7 @@ paths: application/json: schema: type: object - properties: &ref_174 + properties: &ref_172 mime_type: type: string size_in_bytes: @@ -10476,7 +10482,7 @@ paths: application/json: schema: type: object - properties: &ref_175 + properties: &ref_173 msg: type: string content: @@ -10488,7 +10494,7 @@ paths: - Csv - Parquet - Unknown - required: &ref_176 + required: &ref_174 - content_type /w/{workspace}/job_helpers/load_parquet_preview/{path}: get: @@ -10521,7 +10527,11 @@ paths: in: query schema: type: boolean - - name: search + - name: search_col + in: query + schema: + type: string + - name: search_term in: query schema: type: string @@ -10531,34 +10541,6 @@ paths: content: application/json: schema: {} - /w/{workspace}/job_helpers/generate_download_url: - get: - summary: Generate a unique URL to download the file - operationId: generateDownloadUrl - tags: - - helpers - parameters: - - name: workspace - in: path - required: true - schema: *ref_0 - - name: file_key - in: query - required: true - schema: - type: string - responses: - '200': - description: Download URL - content: - application/json: - schema: - type: object - properties: - download_url: - type: string - required: - - download_url /w/{workspace}/job_helpers/delete_s3_file: delete: summary: Permanently delete file from S3 @@ -10608,10 +10590,10 @@ paths: content: application/json: schema: {} - /w/{workspace}/job_helpers/multipart_upload_s3_file: + /w/{workspace}/job_helpers/upload_s3_file: post: - summary: Upload file to S3 bucket using multipart upload - operationId: multipartFileUpload + summary: Upload file to S3 bucket + operationId: fileUpload tags: - helpers parameters: @@ -10619,79 +10601,45 @@ paths: in: path required: true schema: *ref_0 + - name: file_key + in: query + required: false + schema: + type: string + - name: file_extension + in: query + required: false + schema: + type: string + - name: s3_resource_path + in: query + required: false + schema: + type: string requestBody: - description: Query args for a multipart file upload to S3 + description: File content required: true content: - application/json: + application/octet-stream: schema: - type: object - properties: - file_key: - type: string - file_extension: - type: string - part_content: - type: array - items: - type: integer - upload_id: - type: string - parts: - type: array - items: - type: object - properties: &ref_96 - part_number: - type: integer - tag: - type: string - required: &ref_97 - - part_number - - tag - is_final: - type: boolean - cancel_upload: - type: boolean - s3_resource_path: - type: string - file_expiration: - type: string - format: date-time - required: - - part_content - - parts - - is_final - - cancel_upload + type: string + format: binary responses: '200': - description: Chunk upload status + description: File upload status content: application/json: schema: type: object properties: - upload_id: - type: string - parts: - type: array - items: - type: object - properties: *ref_96 - required: *ref_97 - is_done: - type: boolean file_key: type: string required: - file_key - - upload_id - - parts - - is_done - /w/{workspace}/job_helpers/multipart_download_s3_file: - post: + /w/{workspace}/job_helpers/download_s3_file: + get: summary: Download file to S3 bucket - operationId: multipartFileDownload + operationId: fileDownload tags: - helpers parameters: @@ -10699,43 +10647,24 @@ paths: in: path required: true schema: *ref_0 - requestBody: - description: Query args for a multipart file upload to S3 - required: true - content: - application/json: - schema: - type: object - properties: - file_key: - type: string - part_number: - type: integer - file_size: - type: integer - s3_resource_path: - type: string - required: - - file_key - - part_number + - name: file_key + in: query + required: true + schema: + type: string + - name: s3_resource_path + in: query + required: false + schema: + type: string responses: '200': description: Chunk of the downloaded file content: - application/json: + application/octet-stream: schema: - type: object - properties: - file_size: - type: integer - part_content: - type: array - items: - type: integer - next_part_number: - type: integer - required: - - part_content + type: string + format: binary /w/{workspace}/job_metrics/get/{id}: post: summary: get job metrics @@ -10779,46 +10708,46 @@ paths: type: array items: type: object - properties: &ref_177 + properties: &ref_175 id: type: string name: type: string - required: &ref_178 + required: &ref_176 - id scalar_metrics: type: array items: type: object - properties: &ref_179 + properties: &ref_177 metric_id: type: string value: type: number - required: &ref_180 + required: &ref_178 - id - value timeseries_metrics: type: array items: type: object - properties: &ref_181 + properties: &ref_179 metric_id: type: string values: type: array items: type: object - properties: &ref_183 + properties: &ref_181 timestamp: type: string format: date-time value: type: number - required: &ref_184 + required: &ref_182 - timestamp - value - required: &ref_182 + required: &ref_180 - id - values /concurrency_groups/list: @@ -10836,14 +10765,14 @@ paths: type: array items: type: object - properties: &ref_187 + properties: &ref_185 concurrency_id: type: string job_uuids: type: array items: type: string - required: &ref_188 + required: &ref_186 - concurrency_id - job_uuids /concurrency_groups/{concurrency_id}: @@ -10856,7 +10785,7 @@ paths: - name: concurrency_id in: path required: true - schema: &ref_112 + schema: &ref_110 type: string responses: '200': @@ -10890,12 +10819,12 @@ components: name: version in: path required: true - schema: *ref_98 + schema: *ref_96 Token: name: token in: path required: true - schema: *ref_99 + schema: *ref_97 AccountId: name: id in: path @@ -10935,7 +10864,7 @@ components: name: version in: path required: true - schema: *ref_100 + schema: *ref_98 Name: name: name in: path @@ -11043,14 +10972,14 @@ components: filter on created_at for non non started job and started_at otherwise after (exclusive) timestamp in: query - schema: *ref_101 + schema: *ref_99 CreatedOrStartedBefore: name: created_or_started_before description: >- filter on created_at for non non started job and started_at otherwise before (inclusive) timestamp in: query - schema: *ref_102 + schema: *ref_100 Success: name: success description: filter on successful jobs @@ -11065,7 +10994,7 @@ components: name: suspended description: filter on suspended jobs in: query - schema: *ref_103 + schema: *ref_101 Running: name: running description: filter on running jobs @@ -11090,32 +11019,32 @@ components: name: after description: filter on created after (exclusive) timestamp in: query - schema: *ref_104 + schema: *ref_102 Before: name: before description: filter on created before (exclusive) timestamp in: query - schema: *ref_105 + schema: *ref_103 Username: name: username description: filter on exact username of user in: query - schema: *ref_106 + schema: *ref_104 Operation: name: operation description: filter on exact or prefix name of operation in: query - schema: *ref_107 + schema: *ref_105 ResourceName: name: resource description: filter on exact or prefix name of resource in: query - schema: *ref_108 + schema: *ref_106 ActionKind: name: action_kind description: filter on type of operation in: query - schema: *ref_109 + schema: *ref_107 JobKinds: name: job_kinds description: >- @@ -11135,16 +11064,16 @@ components: name: input in: path required: true - schema: *ref_110 + schema: *ref_108 GetStarted: name: get_started in: query - schema: *ref_111 + schema: *ref_109 ConcurrencyId: name: concurrency_id in: path required: true - schema: *ref_112 + schema: *ref_110 schemas: Script: type: object @@ -11155,11 +11084,11 @@ components: properties: *ref_35 required: *ref_36 NewScriptWithDraft: - allOf: *ref_113 + allOf: *ref_111 ScriptHistory: type: object - properties: *ref_114 - required: *ref_115 + properties: *ref_112 + required: *ref_113 ScriptArgs: type: object additionalProperties: *ref_14 @@ -11169,15 +11098,15 @@ components: required: *ref_89 CreateInput: type: object - properties: *ref_116 - required: *ref_117 + properties: *ref_114 + required: *ref_115 UpdateInput: type: object - properties: *ref_118 - required: *ref_119 + properties: *ref_116 + required: *ref_117 RunnableType: type: string - enum: *ref_120 + enum: *ref_118 QueuedJob: type: object properties: *ref_72 @@ -11195,44 +11124,44 @@ components: required: *ref_9 Usage: type: object - properties: *ref_121 + properties: *ref_119 Login: + type: object + properties: *ref_120 + required: *ref_121 + NewUser: type: object properties: *ref_122 required: *ref_123 - NewUser: - type: object - properties: *ref_124 - required: *ref_125 EditWorkspaceUser: type: object - properties: *ref_126 + properties: *ref_124 TruncatedToken: type: object - properties: *ref_127 - required: *ref_128 + properties: *ref_125 + required: *ref_126 NewToken: type: object - properties: *ref_129 + properties: *ref_127 NewTokenImpersonate: type: object - properties: *ref_130 - required: *ref_131 + properties: *ref_128 + required: *ref_129 ListableVariable: type: object properties: *ref_19 required: *ref_20 ContextualVariable: type: object - properties: *ref_132 - required: *ref_133 + properties: *ref_130 + required: *ref_131 CreateVariable: type: object - properties: *ref_134 - required: *ref_135 + properties: *ref_132 + required: *ref_133 EditVariable: type: object - properties: *ref_136 + properties: *ref_134 AuditLog: type: object properties: *ref_1 @@ -11356,44 +11285,44 @@ components: - error Preview: type: object - properties: *ref_137 - required: *ref_138 + properties: *ref_135 + required: *ref_136 CreateResource: type: object - properties: *ref_139 - required: *ref_140 + properties: *ref_137 + required: *ref_138 EditResource: type: object - properties: *ref_141 + properties: *ref_139 Resource: + type: object + properties: *ref_140 + required: *ref_141 + ListableResource: type: object properties: *ref_142 required: *ref_143 - ListableResource: - type: object - properties: *ref_144 - required: *ref_145 ResourceType: type: object properties: *ref_23 required: *ref_24 EditResourceType: type: object - properties: *ref_146 + properties: *ref_144 Schedule: type: object properties: *ref_79 required: *ref_80 ScheduleWJobs: - allOf: *ref_147 + allOf: *ref_145 NewSchedule: + type: object + properties: *ref_146 + required: *ref_147 + EditSchedule: type: object properties: *ref_148 required: *ref_149 - EditSchedule: - type: object - properties: *ref_150 - required: *ref_151 Group: type: object properties: *ref_84 @@ -11407,17 +11336,17 @@ components: properties: *ref_86 required: *ref_87 WorkerPing: + type: object + properties: *ref_150 + required: *ref_151 + UserWorkspaceList: type: object properties: *ref_152 required: *ref_153 - UserWorkspaceList: + CreateWorkspace: type: object properties: *ref_154 required: *ref_155 - CreateWorkspace: - type: object - properties: *ref_156 - required: *ref_157 Workspace: type: object properties: *ref_5 @@ -11434,38 +11363,38 @@ components: allOf: *ref_46 FlowMetadata: type: object - properties: *ref_158 - required: *ref_159 + properties: *ref_156 + required: *ref_157 OpenFlowWPath: allOf: *ref_47 FlowPreview: type: object - properties: *ref_160 - required: *ref_161 + properties: *ref_158 + required: *ref_159 RestartedFrom: type: object - properties: *ref_162 + properties: *ref_160 Policy: type: object properties: *ref_48 ListableApp: type: object - properties: *ref_163 - required: *ref_164 + properties: *ref_161 + required: *ref_162 ListableRawApp: type: object - properties: *ref_165 - required: *ref_166 + properties: *ref_163 + required: *ref_164 AppWithLastVersion: type: object properties: *ref_49 required: *ref_50 AppWithLastVersionWDraft: - allOf: *ref_167 + allOf: *ref_165 AppHistory: type: object - properties: *ref_168 - required: *ref_169 + properties: *ref_166 + required: *ref_167 SlackToken: type: object properties: @@ -11487,8 +11416,8 @@ components: - bot TokenResponse: type: object - properties: *ref_170 - required: *ref_171 + properties: *ref_168 + required: *ref_169 HubScriptKind: name: kind schema: *ref_31 @@ -11501,15 +11430,15 @@ components: properties: *ref_15 WindmillLargeFile: type: object - properties: *ref_172 - required: *ref_173 + properties: *ref_170 + required: *ref_171 WindmillFileMetadata: type: object - properties: *ref_174 + properties: *ref_172 WindmillFilePreview: type: object - properties: *ref_175 - required: *ref_176 + properties: *ref_173 + required: *ref_174 S3Resource: type: object properties: *ref_92 @@ -11520,82 +11449,88 @@ components: required: *ref_17 UploadFilePart: type: object - properties: *ref_96 - required: *ref_97 + properties: + part_number: + type: integer + tag: + type: string + required: + - part_number + - tag MetricMetadata: + type: object + properties: *ref_175 + required: *ref_176 + ScalarMetric: type: object properties: *ref_177 required: *ref_178 - ScalarMetric: + TimeseriesMetric: type: object properties: *ref_179 required: *ref_180 - TimeseriesMetric: + MetricDataPoint: type: object properties: *ref_181 required: *ref_182 - MetricDataPoint: + RawScriptForDependencies: type: object properties: *ref_183 required: *ref_184 - RawScriptForDependencies: + ConcurrencyGroup: type: object properties: *ref_185 required: *ref_186 - ConcurrencyGroup: + StaticTransform: type: object properties: *ref_187 required: *ref_188 - StaticTransform: + JavascriptTransform: type: object properties: *ref_189 required: *ref_190 - JavascriptTransform: - type: object - properties: *ref_191 - required: *ref_192 InputTransform: oneOf: *ref_26 discriminator: *ref_27 RawScript: + type: object + properties: *ref_191 + required: *ref_192 + PathScript: type: object properties: *ref_193 required: *ref_194 - PathScript: + PathFlow: type: object properties: *ref_195 required: *ref_196 - PathFlow: - type: object - properties: *ref_197 - required: *ref_198 FlowModule: type: object properties: *ref_28 required: *ref_29 ForloopFlow: + type: object + properties: *ref_197 + required: *ref_198 + BranchOne: type: object properties: *ref_199 required: *ref_200 - BranchOne: + BranchAll: type: object properties: *ref_201 required: *ref_202 - BranchAll: + Identity: type: object properties: *ref_203 required: *ref_204 - Identity: + Graphql: type: object properties: *ref_205 required: *ref_206 - Graphql: - type: object - properties: *ref_207 - required: *ref_208 FlowModuleValue: - oneOf: *ref_209 - discriminator: *ref_210 + oneOf: *ref_207 + discriminator: *ref_208 Retry: type: object properties: *ref_78 diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 08d574aa57..bfdee0ad09 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -6977,6 +6977,10 @@ paths: in: query schema: type: string + - name: prefix + in: query + schema: + type: string responses: "200": description: List of file keys @@ -7095,7 +7099,7 @@ paths: in: query schema: type: string - + responses: "200": description: Parquet Preview @@ -7103,32 +7107,6 @@ paths: application/json: schema: {} - /w/{workspace}/job_helpers/generate_download_url: - get: - summary: Generate a unique URL to download the file - operationId: generateDownloadUrl - tags: - - helpers - parameters: - - $ref: "#/components/parameters/WorkspaceId" - - name: file_key - in: query - required: true - schema: - type: string - responses: - "200": - description: Download URL - content: - application/json: - schema: - type: object - properties: - download_url: - type: string - required: - - download_url - /w/{workspace}/job_helpers/delete_s3_file: delete: summary: Permanently delete file from S3 @@ -7174,119 +7152,76 @@ paths: application/json: schema: {} - /w/{workspace}/job_helpers/multipart_upload_s3_file: + /w/{workspace}/job_helpers/upload_s3_file: post: - summary: Upload file to S3 bucket using multipart upload - operationId: multipartFileUpload + summary: Upload file to S3 bucket + operationId: fileUpload tags: - helpers parameters: - $ref: "#/components/parameters/WorkspaceId" + - name: file_key + in: query + required: false + schema: + type: string + - name: file_extension + in: query + required: false + schema: + type: string + - name: s3_resource_path + in: query + required: false + schema: + type: string requestBody: - description: Query args for a multipart file upload to S3 + description: File content required: true content: - application/json: + application/octet-stream: schema: - type: object - properties: - file_key: - type: string - file_extension: - type: string - part_content: - type: array - items: - type: integer - upload_id: - type: string - parts: - type: array - items: - $ref: "#/components/schemas/UploadFilePart" - is_final: - type: boolean - cancel_upload: - type: boolean - s3_resource_path: - type: string - file_expiration: - type: string - format: date-time - required: - - part_content - - parts - - is_final - - cancel_upload + type: string + format: binary responses: "200": - description: Chunk upload status + description: File upload status content: application/json: schema: type: object properties: - upload_id: - type: string - parts: - type: array - items: - $ref: "#/components/schemas/UploadFilePart" - is_done: - type: boolean file_key: type: string required: - file_key - - upload_id - - parts - - is_done - /w/{workspace}/job_helpers/multipart_download_s3_file: - post: + /w/{workspace}/job_helpers/download_s3_file: + get: summary: Download file to S3 bucket - operationId: multipartFileDownload + operationId: fileDownload tags: - helpers parameters: - $ref: "#/components/parameters/WorkspaceId" - requestBody: - description: Query args for a multipart file upload to S3 - required: true - content: - application/json: - schema: - type: object - properties: - file_key: - type: string - part_number: - type: integer - file_size: - type: integer - s3_resource_path: - type: string - required: - - file_key - - part_number + - name: file_key + in: query + required: true + schema: + type: string + - name: s3_resource_path + in: query + required: false + schema: + type: string responses: "200": description: Chunk of the downloaded file content: - application/json: + application/octet-stream: schema: - type: object - properties: - file_size: - type: integer - part_content: - type: array - items: - type: integer - next_part_number: - type: integer - required: - - part_content + type: string + format: binary /w/{workspace}/job_metrics/get/{id}: post: diff --git a/backend/windmill-api/src/job_helpers.rs b/backend/windmill-api/src/job_helpers.rs index 2dceea4d97..1c12022c06 100644 --- a/backend/windmill-api/src/job_helpers.rs +++ b/backend/windmill-api/src/job_helpers.rs @@ -1,22 +1,23 @@ +use std::io::{self}; +use std::ops::Range; +use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; -use std::{cmp, time::Duration}; +use std::{cmp, future}; use crate::{db::DB, resources::get_resource_value_interpolated_internal, users::Tokened}; -use anyhow::Context; -use aws_sdk_s3::primitives::DateTime; -use aws_sdk_s3::{ - presigning::PresigningConfig, - primitives::ByteStream, - types::{CompletedMultipartUpload, CompletedPart}, -}; +use axum::body::StreamBody; +use axum::extract::{BodyStream, DefaultBodyLimit}; +use axum::headers::HeaderMap; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; use axum::{ extract::{Path, Query}, routing::{delete, get, post}, Extension, Json, Router, }; +use futures::{StreamExt, TryStreamExt}; use hyper::http; -use itertools::Itertools; -use object_store::ClientConfigKey; +use object_store::{ClientConfigKey, ObjectStore}; use polars::{ io::{ cloud::{AmazonS3ConfigKey, CloudOptions}, @@ -30,13 +31,15 @@ use polars::{ }; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; +use tokio::io::{copy, AsyncWriteExt}; +use tokio_util::io::StreamReader; use tower_http::cors::{Any, CorsLayer}; use windmill_common::error::{Error, JsonResult}; use windmill_common::worker::to_raw_value; use windmill_common::{ db::UserDB, error, - s3_helpers::{build_s3_client, render_endpoint, LargeFileStorage, S3Resource}, + s3_helpers::{build_object_store_client, render_endpoint, LargeFileStorage, S3Resource}, }; use crate::db::ApiAuthed; @@ -85,15 +88,16 @@ pub fn workspaced_service() -> Router { "/load_parquet_preview/*path", get(load_parquet_preview).layer(cors.clone()), ) - .route( - "/generate_download_url", - get(generate_download_url).layer(cors.clone()), - ) .route( "/delete_s3_file", delete(delete_s3_file).layer(cors.clone()), ) .route("/move_s3_file", get(move_s3_file).layer(cors.clone())) + .route("/upload_s3_file", post(upload_s3_file).layer(cors.clone())) + .route( + "/download_s3_file", + get(download_s3_file).layer(cors.clone()), + ) .route( "/multipart_upload_s3_file", post(multipart_upload_s3_file).layer(cors.clone()), @@ -102,6 +106,7 @@ pub fn workspaced_service() -> Router { "/multipart_download_s3_file", post(multipart_download_s3_file).layer(cors.clone()), ) + .layer(DefaultBodyLimit::max(100 * 1024 * 1024)) // necessary for multipart upload } #[derive(Deserialize)] @@ -116,19 +121,19 @@ struct DuckdbConnectionSettingsResponse { async fn duckdb_connection_settings( Path(_w_id): Path, Json(query): Json, -) -> error::JsonResult { +) -> JsonResult { let mut duckdb_settings: String = String::new(); let s3_resource = query.s3_resource; - duckdb_settings.push_str(format!("SET home_directory='./';\n").as_str()); // TODO: make this configurable maybe, or point to a temporary folder - duckdb_settings.push_str(format!("INSTALL 'httpfs';\n").as_str()); + duckdb_settings.push_str("SET home_directory='./';\n"); // TODO: make this configurable maybe, or point to a temporary folder + duckdb_settings.push_str("INSTALL 'httpfs';\n"); if s3_resource.path_style { - duckdb_settings.push_str(format!("SET s3_url_style='path';\n").as_str()); + duckdb_settings.push_str("SET s3_url_style='path';\n"); } duckdb_settings.push_str(format!("SET s3_region='{}';\n", s3_resource.region).as_str()); duckdb_settings.push_str(format!("SET s3_endpoint='{}';\n", s3_resource.endpoint).as_str()); if !s3_resource.use_ssl { - duckdb_settings.push_str(format!("SET s3_use_ssl=0;\n").as_str()); // default is true for DuckDB + duckdb_settings.push_str("SET s3_use_ssl=0;\n"); // default is true for DuckDB } if let Some(access_key_id) = s3_resource.access_key { duckdb_settings.push_str(format!("SET s3_access_key_id='{}';\n", access_key_id).as_str()); @@ -154,7 +159,7 @@ async fn duckdb_connection_settings_v2( Tokened { token }: Tokened, Path(w_id): Path, Json(query): Json, -) -> error::JsonResult { +) -> JsonResult { let s3_resource_opt = match query.s3_resource_path { Some(s3_resource_path) => { get_s3_resource( @@ -173,7 +178,7 @@ async fn duckdb_connection_settings_v2( s3_resource_opt } }; - let s3_resource = s3_resource_opt.ok_or(error::Error::NotFound( + let s3_resource = s3_resource_opt.ok_or(Error::NotFound( "No datasets storage resource defined at the workspace level".to_string(), ))?; return duckdb_connection_settings( @@ -196,7 +201,7 @@ struct PolarsConnectionSettings { async fn polars_connection_settings( Path(_w_id): Path, Json(query): Json, -) -> error::JsonResult { +) -> JsonResult { let s3_resource = query.s3_resource; let response = S3fsArgs { @@ -251,7 +256,7 @@ async fn polars_connection_settings_v2( Tokened { token }: Tokened, Path(w_id): Path, Json(query): Json, -) -> error::JsonResult { +) -> JsonResult { let s3_resource_opt = match query.s3_resource_path { Some(s3_resource_path) => { get_s3_resource( @@ -270,7 +275,7 @@ async fn polars_connection_settings_v2( s3_resource_opt } }; - let s3_resource = s3_resource_opt.ok_or(error::Error::NotFound( + let s3_resource = s3_resource_opt.ok_or(Error::NotFound( "No datasets storage resource defined at the workspace level".to_string(), ))?; let s3fs = polars_connection_settings( @@ -304,7 +309,7 @@ async fn s3_resource_info( Tokened { token }: Tokened, Path(w_id): Path, Json(query): Json, -) -> error::JsonResult { +) -> JsonResult { let s3_resource_opt = match query.s3_resource_path { Some(s3_resource_path) => { get_s3_resource( @@ -323,7 +328,7 @@ async fn s3_resource_info( s3_resource_opt } }; - let s3_resource = s3_resource_opt.ok_or(error::Error::NotFound( + let s3_resource = s3_resource_opt.ok_or(Error::NotFound( "No datasets storage resource defined at the workspace level".to_string(), ))?; return Ok(Json(s3_resource)); @@ -339,35 +344,35 @@ async fn test_connection( Extension(db): Extension, Tokened { token }: Tokened, Path(w_id): Path, -) -> error::JsonResult<()> { +) -> JsonResult<()> { let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; if s3_resource_opt.is_none() { - return Err(error::Error::NotFound( + return Err(Error::NotFound( "No datasets storage resource defined at the workspace level".to_string(), )); } - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let s3_client = build_s3_client(&s3_resource); + let s3_client = build_object_store_client(&s3_resource)?; s3_client - .list_objects() - .bucket(s3_resource.bucket) - .max_keys(1) - .send() + .list(None) + .next() .await + .transpose() .map_err(|err| { tracing::error!("Error testing connection to S3 bucket: {:?}", err); - error::Error::InternalErr(err.to_string()) + Error::InternalErr(format!("Error testing connection: {}", err.to_string())) })?; return Ok(Json(())); } #[derive(Deserialize)] struct ListStoredFilesQuery { - pub max_keys: i32, + pub max_keys: usize, pub marker: Option, + pub prefix: Option, } #[derive(Serialize)] @@ -384,7 +389,7 @@ async fn list_stored_files( Tokened { token }: Tokened, Path(w_id): Path, Query(query): Query, -) -> error::JsonResult { +) -> JsonResult { let (public_resource, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, Some(user_db), &token, &w_id).await?; if !public_resource.unwrap_or(false) && s3_resource_opt.is_none() { @@ -395,54 +400,40 @@ async fn list_stored_files( })); } - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let s3_client = build_s3_client(&s3_resource); + let s3_client = build_object_store_client(&s3_resource)?; - let s3_bucket = s3_resource.bucket; + let prefix = query.prefix.map(object_store::path::Path::from); + let object_stream = if let Some(marker) = query.marker { + s3_client.list_with_offset(prefix.as_ref(), &object_store::path::Path::from(marker)) + } else { + s3_client.list(prefix.as_ref()) + }; - let list_object_query = s3_client - .list_objects() - .bucket(s3_bucket.clone()) - .max_keys(query.max_keys) - .set_marker(query.marker); - - let bucket_objects = list_object_query - .send() + let stored_datasets = object_stream + .take((query.max_keys) as usize) + .map(|obj| obj.map(|x| WindmillLargeFile { s3: x.location.to_string() })) + .try_collect::>() .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|e| Error::InternalErr(format!("Error listring files: {e}",)))?; - let stored_datasets = bucket_objects - .contents() - .iter() - .filter(|object| object.key().is_some()) - .map(|object| object.key()) - .map(Option::unwrap) - .map(&str::to_string) - .map(|object_key| WindmillLargeFile { s3: object_key.clone() }) - .collect::>(); + let next_marker = if stored_datasets.len() > query.max_keys as usize { + Some((stored_datasets.last().unwrap().s3).clone()) + } else { + None + }; + // TODO: ideally do it on the size hint, if reliable #[cfg(not(feature = "enterprise"))] if stored_datasets.len() > 20 { - return Err(error::Error::ExecutionErr( + return Err(Error::ExecutionErr( "The workspace s3 bucket contains more than 20 files. Consider upgrading to Windmill Enterprise Edition to continue to use this feature." .to_string(), )); } - let next_marker = if bucket_objects.is_truncated().unwrap_or(false) { - if bucket_objects.next_marker().is_some() { - // some S3 providers returns the next marker for us. If that's the case just re-use it - bucket_objects.next_marker().map(|v| v.to_owned()) - } else { - // others, like AWS, doesn't and implicitly expect users to return the last key of the current page - stored_datasets.last().map(|v| v.s3.clone()) - } - } else { - None - }; - return Ok(Json(ListStoredDatasetsResponse { windmill_large_files: stored_datasets, next_marker, @@ -469,7 +460,7 @@ struct LoadFilePreviewQuery { pub file_key: String, // The two options below are requested from s3 with an additional query is not set - pub file_size_in_bytes: Option, + pub file_size_in_bytes: Option, pub file_mime_type: Option, // For CSV files, the separator needs to be specified @@ -480,8 +471,8 @@ struct LoadFilePreviewQuery { // - CSVs: only the length will be taken into account, a CSV file larger than this will be truncated. // Note that truncated CSV files might not be valid CSV files anymore, and therefore the preview might fail // - Parquet files: Parquet files are lazy-loaded. Therefore none of those params will be taken into account - pub read_bytes_from: i64, - pub read_bytes_length: i64, + pub read_bytes_from: usize, + pub read_bytes_length: usize, } #[derive(Serialize)] @@ -505,36 +496,27 @@ async fn load_file_metadata( Tokened { token }: Tokened, Path(w_id): Path, Query(query): Query, -) -> error::JsonResult { +) -> JsonResult { let file_key = query.file_key.clone(); let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let s3_client = build_s3_client(&s3_resource); + let s3_client = build_object_store_client(&s3_resource)?; - let s3_bucket = s3_resource.bucket.clone(); - let s3_object_metadata = s3_client - .head_object() - .bucket(&s3_bucket) - .key(&file_key) - .send() - .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + let path = object_store::path::Path::from(file_key.as_str()); + let s3_object_metadata = s3_client.head(&path).await.map_err(|err| { + tracing::error!("Error loading file metadata: {:?}", err); + Error::InternalErr(format!("Error loading file metadata: {}", err.to_string())) + })?; let response = LoadFileMetadataResponse { - mime_type: s3_object_metadata.content_type().map(&str::to_string), - size_in_bytes: s3_object_metadata.content_length(), - last_modified: s3_object_metadata - .last_modified() - .map(|dt| chrono::DateTime::from_timestamp(dt.secs(), dt.subsec_nanos())) - .flatten(), - expires: s3_object_metadata - .expires() - .map(|dt| chrono::DateTime::from_timestamp(dt.secs(), dt.subsec_nanos())) - .flatten(), - version_id: s3_object_metadata.version_id().map(&str::to_string), + mime_type: None, + size_in_bytes: Some(s3_object_metadata.size as i64), + last_modified: Some(s3_object_metadata.last_modified.to_utc()), + expires: None, + version_id: s3_object_metadata.version, }; return Ok(Json(response)); } @@ -555,10 +537,10 @@ async fn load_parquet_preview( Tokened { token }: Tokened, Path((w_id, file_key)): Path<(String, String)>, Query(query): Query, -) -> error::JsonResult> { +) -> JsonResult> { let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; @@ -584,10 +566,10 @@ async fn load_file_preview( Tokened { token }: Tokened, Path(w_id): Path, Query(query): Query, -) -> error::JsonResult { +) -> JsonResult { // query validation if query.read_bytes_length > 8 * 1024 * 1024 { - return Err(error::Error::BadRequest( + return Err(Error::BadRequest( "Cannot load file bigger than 8MB".to_string(), )); } @@ -595,29 +577,22 @@ async fn load_file_preview( let file_key = query.file_key.clone(); let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let s3_client = build_s3_client(&s3_resource); - - let s3_bucket = s3_resource.bucket.clone(); + let s3_client = build_object_store_client(&s3_resource)?; // if content length is provided in the request, use it, otherwise get it from s3 - let (s3_object_mime_type, s3_object_content_length) = + let s3_object_content_length = if query.file_size_in_bytes.is_some() || query.file_mime_type.is_some() { - (query.file_mime_type.clone(), query.file_size_in_bytes) + query.file_size_in_bytes } else { - let s3_object_metadata = s3_client - .head_object() - .bucket(&s3_bucket) - .key(&file_key) - .send() - .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; - ( - s3_object_metadata.content_type().map(|v| v.to_owned()), - s3_object_metadata.content_length(), - ) + let path = object_store::path::Path::from(file_key.clone()); + let s3_object_metadata = s3_client.head(&path).await.map_err(|err| { + tracing::error!("Error loading file preview: {:?}", err); + Error::InternalErr(format!("Error loading file preview: {}", err.to_string())) + })?; + Some(s3_object_metadata.size) }; let file_chunk_length = if s3_object_content_length.is_some() { @@ -629,81 +604,43 @@ async fn load_file_preview( query.read_bytes_length }; let content_type: WindmillContentType; - let content_preview = match s3_object_mime_type.as_deref() { - Some("text/csv") => { + + let lowercased_file_key = file_key.clone().to_lowercase(); + let content_preview = + if lowercased_file_key.ends_with(".csv") || lowercased_file_key.ends_with("tsv") { content_type = WindmillContentType::Csv; csv_file_preview_with_fallback( - &s3_client, - &s3_bucket, + s3_client, &file_key, file_chunk_length, query.csv_separator, query.csv_has_header, ) .await - } - Some(mt) - if mt.starts_with("text/") - || mt == "application/json" - || mt == "application/x-yaml" => + } else if lowercased_file_key.ends_with(".json") + || lowercased_file_key.ends_with(".yaml") + || lowercased_file_key.ends_with(".yml") + || lowercased_file_key.ends_with(".xml") + || lowercased_file_key.ends_with(".txt") + || lowercased_file_key.ends_with(".log") { content_type = WindmillContentType::RawText; read_s3_text_object_head( - &s3_client, - &s3_bucket, + s3_client, &file_key, query.read_bytes_from, file_chunk_length, ) .await - } - mt_opt => { - let lowercased_file_key = file_key.to_lowercase(); - // sometimes S3 doesn't infer the content type on upload. Guess it from the file extension - if lowercased_file_key.ends_with(".parquet") { - content_type = WindmillContentType::Parquet; - read_s3_parquet_object_head(&s3_resource, &file_key).await - } else if lowercased_file_key.ends_with(".csv") { - content_type = WindmillContentType::Csv; - csv_file_preview_with_fallback( - &s3_client, - &s3_bucket, - &file_key, - file_chunk_length, - query.csv_separator, - query.csv_has_header, - ) - .await - } else if lowercased_file_key.ends_with(".json") - || lowercased_file_key.ends_with(".yaml") - || lowercased_file_key.ends_with(".yml") - || lowercased_file_key.ends_with(".xml") - || lowercased_file_key.ends_with(".txt") - || lowercased_file_key.ends_with(".log") - { - content_type = WindmillContentType::RawText; - read_s3_text_object_head( - &s3_client, - &s3_bucket, - &file_key, - query.read_bytes_from, - file_chunk_length, - ) - .await - } else { - content_type = WindmillContentType::Unknown; - let msg = match mt_opt { - Some(mt) => { - format!("Preview is not available for content of type '{}'", mt).to_string() - } - None => "Preview is not available. Content type is unknown or not supported" - .to_string(), - }; - Err(error::Error::ExecutionErr(msg)) - } - } - }; - + } else if lowercased_file_key.ends_with(".parquet") { + content_type = WindmillContentType::Parquet; + read_s3_parquet_object_head(&s3_resource, &file_key).await + } else { + content_type = WindmillContentType::Unknown; + Err(Error::ExecutionErr( + "Preview is not available. Content type is unknown or not supported".to_string(), + )) + }; let response: LoadFilePreviewResponse = match content_preview { Ok(content) => LoadFilePreviewResponse { content_type, content: Some(content), msg: None }, @@ -711,53 +648,9 @@ async fn load_file_preview( LoadFilePreviewResponse { content_type, content: None, msg: Some(err.to_string()) } } }; - return Ok(Json(response)); } -#[derive(Deserialize)] -struct GenerateDownloadUrlQuery { - pub file_key: String, -} - -#[derive(Serialize)] -struct GenerateDownloadUrlResponse { - pub download_url: String, -} - -async fn generate_download_url( - authed: ApiAuthed, - Extension(db): Extension, - Tokened { token }: Tokened, - Path(w_id): Path, - Query(query): Query, -) -> JsonResult { - let file_key = query.file_key.clone(); - let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; - - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( - "No files storage resource defined at the workspace level".to_string(), - ))?; - - let s3_client = build_s3_client(&s3_resource); - let s3_bucket = s3_resource.bucket.clone(); - - // URL expires 5 minutes after its generation - let presigned_config = PresigningConfig::expires_in(Duration::from_secs(60 * 5)) - .map_err(|err| error::Error::InternalErr(err.to_string()))?; - let download_url = s3_client - .get_object() - .bucket(&s3_bucket) - .key(&file_key) - .presigned(presigned_config) - .await - .map_err(|err| error::Error::InternalErr(err.to_string()))? - .uri() - .to_string(); - - return Ok(Json(GenerateDownloadUrlResponse { download_url })); -} - #[derive(Deserialize)] struct DeleteS3FileQuery { pub file_key: String, @@ -769,26 +662,21 @@ async fn delete_s3_file( Tokened { token }: Tokened, Path(w_id): Path, Query(query): Query, -) -> error::JsonResult<()> { +) -> JsonResult<()> { let file_key = query.file_key.clone(); let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let s3_client = build_s3_client(&s3_resource); + let s3_client = build_object_store_client(&s3_resource)?; - let s3_bucket = s3_resource.bucket.clone(); - s3_client - .delete_object() - .bucket(&s3_bucket) - .key(&file_key) - .send() - .await - .map_err(|err| { - tracing::error!("{:?}", err); - error::Error::InternalErr(err.to_string()) - })?; + let path = object_store::path::Path::from(file_key.as_str()); + + s3_client.delete(&path).await.map_err(|err| { + tracing::error!("Error deleting file: {:?}", err); + Error::InternalErr(format!("Error deleting file: {}", err.to_string())) + })?; return Ok(Json(())); } @@ -804,66 +692,60 @@ async fn move_s3_file( Tokened { token }: Tokened, Path(w_id): Path, Query(query): Query, -) -> error::JsonResult<()> { +) -> JsonResult<()> { let (_, s3_resource_opt) = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?; - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let s3_client = build_s3_client(&s3_resource); + let s3_client = build_object_store_client(&s3_resource)?; - let s3_bucket = s3_resource.bucket.clone(); - let source_uri = format!("{}/{}", s3_bucket, query.src_file_key); - s3_client - .copy_object() - .copy_source(&source_uri) - .bucket(&s3_bucket) - .key(&query.dest_file_key) - .send() - .await - .map_err(|err| { - tracing::error!("{:?}", err); - error::Error::InternalErr(err.to_string()) - })?; + let source_path = object_store::path::Path::from(query.src_file_key); + let dest_path = object_store::path::Path::from(query.dest_file_key); s3_client - .delete_object() - .bucket(&s3_bucket) - .key(&query.src_file_key) - .send() + .copy(&source_path, &dest_path) .await .map_err(|err| { - tracing::error!("{:?}", err); - error::Error::InternalErr(err.to_string()) + tracing::error!("Error copying object: {:?}", err); + Error::InternalErr(format!("Error copying object: {}", err.to_string())) })?; + + s3_client.delete(&source_path).await.map_err(|err| { + tracing::error!("Error removing object post copy: {:?}", err); + Error::InternalErr(format!( + "Error removing object post copy: {}", + err.to_string() + )) + })?; return Ok(Json(())); } #[derive(Deserialize)] struct DownloadFileQuery { pub file_key: String, - - pub part_number: i64, // part number of the file to download. A file part are approx 5MB - pub file_size: Option, // leave empty for the first call, it will be returned in the response and ideally sent back in the next call - pub s3_resource_path: Option, } -#[derive(Serialize)] -struct DownloadFileResponse { - pub file_size: Option, - pub part_content: Vec, - pub next_part_number: Option, // is None when this is the last part being returned +async fn multipart_download_s3_file( + _authed: ApiAuthed, + Extension(_db): Extension, + Extension(_user_db): Extension, + Tokened { token: _token }: Tokened, + Path(_w_id): Path, + Json(_value): Json, +) -> JsonResult<()> { + return Err(Error::BadRequest("This endpoint has been replaced by /download_s3_file. Please upgrade your client SDK to the latest version and migrate your code to the latest endpoints.".to_string())); } -async fn multipart_download_s3_file( +async fn download_s3_file( authed: ApiAuthed, Extension(db): Extension, Extension(user_db): Extension, Tokened { token }: Tokened, Path(w_id): Path, - Json(query): Json, -) -> error::JsonResult { + Query(query): Query, +) -> error::Result { let s3_resource_opt = match query.s3_resource_path.clone() { Some(s3_resource_path) => { get_s3_resource( @@ -883,66 +765,27 @@ async fn multipart_download_s3_file( } }; - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let bucket = s3_resource.bucket.clone(); - let file_key = query.file_key.clone(); - let s3_client = build_s3_client(&s3_resource); + let s3_client = build_object_store_client(&s3_resource)?; - let file_size = match query.file_size { - Some(fs) => Some(fs), - None => { - let s3_object_metadata = s3_client - .head_object() - .bucket(&bucket) - .key(&file_key) - .send() - .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; - s3_object_metadata.content_length() - } - }; - - let chunk_size_bytes: i64 = 5 * 1024 * 1024; - let from_byte = query.part_number * chunk_size_bytes; - let (length, next_part_number) = - if file_size.is_some() && file_size.unwrap() - from_byte <= chunk_size_bytes { - (file_size.unwrap() - from_byte, None) - } else { - (chunk_size_bytes, Some(query.part_number + 1)) - }; - - let payload = - read_object_chunk(&s3_client, bucket.as_str(), &file_key, from_byte, length).await?; - tracing::warn!( - "Reading chunk {} with length {} - payload size: {}", - from_byte, - length, - payload.len() - ); - - return Ok(Json(DownloadFileResponse { - file_size, - part_content: payload, - next_part_number, - })); + let path = object_store::path::Path::from(query.file_key); + let s3_object = s3_client.get(&path).await.map_err(|err| { + tracing::warn!("Error retrieving file from S3: {:?}", err); + Error::InternalErr(format!("Error retrieving file: {}", err.to_string())) + })?; + let body_stream = StreamBody::new(s3_object.into_stream()); + let mut headers = HeaderMap::new(); + headers.insert("content-type", "application/octet-stream".parse().unwrap()); + return Ok((StatusCode::OK, headers, body_stream).into_response()); } #[derive(Deserialize)] struct UploadFileQuery { pub file_key: Option, // if none, the file will be placed in windmill_uploads/ with a random name. pub file_extension: Option, // preferred extension for the file in case a random name has to be generated - pub part_content: Vec, - pub upload_id: Option, // should be None for the first call to initiate the upload - - pub parts: Vec, // parts already uploaded, with their part_number and the tag associated - - pub is_final: bool, // whether it's the final chunk - pub cancel_upload: bool, // whether the upload should be cancelled. upload_id should be set. subsequent calls with this upload_id will fail - pub s3_resource_path: Option, // custom S3 resource to use for this upload. It None, the workspace S3 resource will be used - pub file_expiration: Option>, } #[derive(Deserialize, Serialize, Clone)] @@ -953,25 +796,91 @@ struct UploadFilePart { #[derive(Serialize)] struct UploadFileResponse { - pub upload_id: String, - pub parts: Vec, // parts already uploaded, with their part_number and the tag associated - pub is_done: bool, // whether the transfer is finished, either b/c it got cancelled or because the last chunk was uploaded pub file_key: String, } +use std::io::Result; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::Duration; + +use pin_project::pin_project; +use tokio::io::{AsyncRead, ReadBuf}; +use tokio::time::{interval, Interval}; + +#[pin_project] +pub struct ProgressReadAdapter { + #[pin] + inner: R, + interval: Interval, + interval_bytes: usize, + total_bytes: usize, +} + +impl ProgressReadAdapter { + pub fn new(inner: R) -> Self { + Self { + inner, + interval: interval(Duration::from_millis(100)), + interval_bytes: 0, + total_bytes: 0, + } + } +} + +impl AsyncRead for ProgressReadAdapter { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.project(); + + let before = buf.filled().len(); + let result = this.inner.poll_read(cx, buf); + let after = buf.filled().len(); + + *this.interval_bytes += after - before; + match this.interval.poll_tick(cx) { + Poll::Pending => {} + Poll::Ready(_) => { + tracing::info!("read {} bytes for s3 upload", *this.interval_bytes * 10); + *this.interval_bytes = 0; + } + }; + *this.total_bytes += *this.interval_bytes; + + #[cfg(not(feature = "enterprise"))] + if *this.total_bytes > 50 * 1024 * 1024 { + let err = error::Error::BadRequest( + "Uploading files bigger than 50Mb is only permitted in Windmill EE".to_string(), + ); + return Poll::Ready(Err(io::Error::new(io::ErrorKind::ConnectionAborted, err))); + } + result + } +} + async fn multipart_upload_s3_file( + _authed: ApiAuthed, + Extension(_db): Extension, + Extension(_user_db): Extension, + Tokened { token: _token }: Tokened, + Path(_w_id): Path, + Json(_value): Json, +) -> JsonResult<()> { + return Err(Error::BadRequest("This endpoint has been replaced by /upload_s3_file. Please upgrade your client SDK to the latest version to migrate your code.".to_string())); +} + +async fn upload_s3_file( authed: ApiAuthed, Extension(db): Extension, Extension(user_db): Extension, Tokened { token }: Tokened, Path(w_id): Path, - Json(query): Json, -) -> error::JsonResult { - tracing::debug!( - "Multi part file upload: part number: {} - is_final: {}", - query.parts.len(), - query.is_final - ); + Query(query): Query, + body: BodyStream, +) -> JsonResult { let file_key = match query.file_key.clone() { Some(fk) => fk, None => { @@ -1009,121 +918,50 @@ async fn multipart_upload_s3_file( } }; - let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr( + let s3_resource = s3_resource_opt.ok_or(Error::InternalErr( "No files storage resource defined at the workspace level".to_string(), ))?; - let bucket = s3_resource.bucket.clone(); - let s3_client = build_s3_client(&s3_resource); + let s3_client = build_object_store_client(&s3_resource)?; - if query.cancel_upload && query.upload_id.clone().is_some() { - let upload_id = query.upload_id.unwrap(); - s3_client - .abort_multipart_upload() - .bucket(&bucket) - .key(&file_key) - .upload_id(&upload_id) - .send() - .await - .map_err(|err| { - tracing::error!("{:?}", err); - error::Error::InternalErr(err.to_string()) - })?; - return Ok(Json(UploadFileResponse { - upload_id, - parts: vec![], // empty parts as the transfer has been cancelled - is_done: true, - file_key, - })); - } + let body_with_io_error = body.map_err(|err| io::Error::new(io::ErrorKind::Other, err)); + let body_reader = StreamReader::new(body_with_io_error); + futures::pin_mut!(body_reader); - let (upload_id, part_number) = match query { - UploadFileQuery { upload_id: Some(upload_id), ref parts, .. } if parts.len() > 0 => { - (upload_id, parts.len() + 1) - } - UploadFileQuery { upload_id: None, ref parts, .. } if parts.len() == 0 => { - let mut upload_builder = s3_client - .create_multipart_upload() - .bucket(&bucket) - .key(&file_key); - if let Some(file_expiration) = query.file_expiration { - upload_builder = upload_builder.expires(DateTime::from_secs(file_expiration.timestamp())); - } - let multipart_upload_res = upload_builder - .send() - .await - .map_err(|err| { - tracing::error!("{:?}", err); - error::Error::InternalErr(err.to_string()) - })?; - let upload_id = multipart_upload_res - .upload_id - .context("Upload ID is missing in the response")?; - (upload_id, 1) - } - _ => { - return Err(error::Error::BadRequest( - "parts should be empty when upload_id is not provided, as a new upload will be created." - .to_string(), - )) - } - }; + let path = object_store::path::Path::from(file_key.clone()); + let (multipart_id, mut parts_writer) = s3_client.put_multipart(&path).await.map_err(|err| { + tracing::error!("Error initializing multipart upload: {:?}", err); + Error::InternalErr(format!( + "Error initializing multipart upload: {}", + err.to_string() + )) + })?; + let mut progressed_body_reader = ProgressReadAdapter::new(&mut body_reader); - let chunk_content = query.part_content.clone(); - let chunk_content_stream = ByteStream::from(chunk_content); - let multipart_upload_res = s3_client - .upload_part() - .bucket(&bucket) - .key(&file_key) - .upload_id(&upload_id) - .body(chunk_content_stream) - .part_number(part_number as i32) - .send() + copy(&mut progressed_body_reader, &mut parts_writer) .await .map_err(|err| { - tracing::error!("{:?}", err); - error::Error::InternalErr(err.to_string()) + let _ = s3_client.abort_multipart(&path, &multipart_id); + tracing::error!("Error forwarding stream to object writer: {:?}", err); + Error::InternalErr(format!("Error copying stream: {}", err.to_string())) })?; - let mut new_parts = query.parts.clone(); - new_parts.push(UploadFilePart { - part_number: part_number as u16, - tag: multipart_upload_res.e_tag.unwrap_or_default(), - }); - - if query.is_final { - let completed_parts = new_parts - .iter() - .map(|part| { - CompletedPart::builder() - .e_tag(&part.tag) - .part_number(part.part_number as i32) - .build() - }) - .collect_vec(); - let _complete_multipart_upload_res = s3_client - .complete_multipart_upload() - .bucket(&s3_resource.bucket) - .key(&file_key) - .upload_id(&upload_id) - .multipart_upload( - CompletedMultipartUpload::builder() - .set_parts(Some(completed_parts)) - .build(), - ) - .send() - .await - .map_err(|err| { - tracing::error!("{:?}", err); - error::Error::InternalErr(err.to_string()) - })?; - } - - return Ok(Json(UploadFileResponse { - upload_id, - parts: new_parts, - is_done: query.is_final, - file_key, - })); + parts_writer.flush().await.map_err(|err| { + let _ = s3_client.abort_multipart(&path, &multipart_id); + tracing::error!("Error flushing multipart writer: {:?}", err); + Error::InternalErr(format!( + "Error flushing multipart writer: {}", + err.to_string() + )) + })?; + parts_writer.shutdown().await.map_err(|err| { + let _ = s3_client.abort_multipart(&path, &multipart_id); + tracing::error!("Error finishing multipart upload: {:?}", err); + Error::InternalErr(format!( + "Error finishing multipart upload: {}", + err.to_string() + )) + })?; + return Ok(Json(UploadFileResponse { file_key })); } #[derive(Deserialize)] @@ -1158,7 +996,7 @@ async fn get_workspace_s3_resource<'c>( "Value stored in large_file_storage column is invalid and could not be deserialized: {}", err ); - error::Error::InternalErr( + Error::InternalErr( "Could not deserialize LargeFileStorage value found in database".to_string(), ) })?; @@ -1215,11 +1053,14 @@ async fn get_s3_resource<'c>( .await?; if s3_resource_value_raw.is_none() { - return Err(error::Error::NotFound("Resource not found".to_string())); + return Err(Error::NotFound("Resource not found".to_string())); } let s3_resource = serde_json::from_value::(s3_resource_value_raw.unwrap()) - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|err| { + tracing::error!("Error deserializing S3 resource: {:?}", err); + Error::InternalErr(format!("Error reading s3 resource: {}", err.to_string())) + })?; return Ok(Some(s3_resource)); } @@ -1251,56 +1092,37 @@ fn build_polars_s3_config(s3_resource_ref: &S3Resource) -> CloudOptions { } async fn read_object_chunk( - s3_client: &aws_sdk_s3::Client, - s3_bucket: &str, + s3_client: Arc, file_key: &str, - from_byte: i64, - length: i64, + from_byte: usize, + length: usize, ) -> error::Result> { + let path = object_store::path::Path::from(file_key); let s3_object = s3_client - .get_object() - .range(format!("bytes={}-{}", from_byte, from_byte + length - 1).to_string()) - .bucket(s3_bucket) - .key(file_key) - .send() + .get_range(&path, Range { start: from_byte, end: from_byte + length }) .await .map_err(|err| { tracing::warn!("Error fetching text file from S3: {:?}", err); - error::Error::InternalErr(err.to_string()) + Error::InternalErr(err.to_string()) })?; - let payload = s3_object - .body - .collect() - .await - .map_err(|err| { - tracing::warn!( - "Error reading raw text file {}. Error was: {:?}", - file_key, - err - ); - error::Error::InternalErr("File encoding is not supported".to_string()) - })? - .into_bytes() - .to_vec(); - return Ok(payload); + return Ok(s3_object.to_vec()); } async fn read_s3_text_object_head( - s3_client: &aws_sdk_s3::Client, - s3_bucket: &str, + s3_client: Arc, file_key: &str, - from_byte: i64, - length: i64, + from_byte: usize, + length: usize, ) -> error::Result { - let payload = read_object_chunk(s3_client, s3_bucket, file_key, from_byte, length).await?; + let payload = read_object_chunk(s3_client, file_key, from_byte, length).await?; let file_header_str = String::from_utf8(payload).map_err(|err| { tracing::warn!( "Encoding of file {} unsupported. Error was: {:?}", file_key, err ); - error::Error::InternalErr("File encoding is not supported".to_string()) + Error::InternalErr("File encoding is not supported".to_string()) })?; return Ok(file_header_str); } @@ -1331,20 +1153,22 @@ async fn read_s3_parquet_object_head( match lzdf_result { Err(err) => { tracing::warn!("Error fetching parquet file from S3: {:?}", err); - return Err(error::Error::InternalErr(err.to_string())); + return Err(Error::InternalErr(err.to_string())); } Ok(lzdf) => { let df = lzdf .select(&[col("*")]) .limit(10) // for now read only first 10 lines .collect() - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|err| { + Error::InternalErr(format!("Error querying dataset: {}", err.to_string())) + })?; return Ok(format!("{:?}", df).to_string()); } } }) .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|err| Error::InternalErr(format!("Error reading dataset: {}", err.to_string())))?; return polars_df_result; } @@ -1379,7 +1203,7 @@ async fn read_s3_parquet_chunk( match lzdf_result { Err(err) => { tracing::warn!("Error fetching parquet file from S3: {:?}", err); - return Err(error::Error::InternalErr(err.to_string())); + return Err(Error::InternalErr(err.to_string())); } Ok(lzdf) => { let df = lzdf @@ -1405,82 +1229,85 @@ async fn read_s3_parquet_chunk( } else { df }; - let df = df - .collect() - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + let df = df.collect().map_err(|err| { + Error::InternalErr(format!("Error collecting dataset: {}", err.to_string())) + })?; return Ok(to_raw_value(&df)); } } }) .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|err| { + tracing::error!("Error reading dataset: {:?}", err); + Error::InternalErr(format!("Error reading dataset: {}", err.to_string())) + })?; } async fn csv_file_preview_with_fallback( - s3_client: &aws_sdk_s3::Client, - s3_bucket: &str, + s3_client: Arc, file_key: &str, - length: i64, + length: usize, separator: Option, has_header: Option, ) -> error::Result { - match read_s3_csv_object_head( - &s3_client, &s3_bucket, &file_key, length, separator, has_header, - ) - .await + match read_s3_csv_object_head(s3_client.clone(), &file_key, length, separator, has_header).await { Ok(csv_preview) => Ok(csv_preview), Err(_) => { // fallback to default text file preview is the CSV could not be parsed. It's a text file after all let raw_text = - read_s3_text_object_head(&s3_client, &s3_bucket, &file_key, 0, length).await?; + read_s3_text_object_head(s3_client.clone(), &file_key, 0, length).await?; return Ok(raw_text); } } } async fn read_s3_csv_object_head( - s3_client: &aws_sdk_s3::Client, - s3_bucket: &str, + s3_client: Arc, file_key: &str, - length: i64, + length: usize, separator: Option, has_header: Option, ) -> error::Result { let separator_final = match separator { Some(separator_char) if separator_char == "\\t" => Ok("\t".as_bytes()[0]), - Some(separator_char) if separator_char.len() != 1 => Err(error::Error::BadRequest( + Some(separator_char) if separator_char.len() != 1 => Err(Error::BadRequest( "Separator must be a single character".to_string(), )), Some(separator_char) => Ok(separator_char.as_bytes()[0]), None => Ok(",".as_bytes()[0]), // polars uses the comma as default, doing the same here }?; + let path = object_store::path::Path::from(file_key); let s3_object = s3_client - .get_object() - .bucket(s3_bucket) - .range(format!("bytes=0-{}", length).to_string()) - .key(file_key) - .send() + .get(&path) .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|err| { + tracing::error!("Error fetching CSV file: {:?}", err); + Error::InternalErr(format!("Error fetching CSV file: {}", err.to_string())) + })? + .into_stream(); // TODO: polars does not seem to support lazy csv reader, unfortunately. We can implement it ourselves if needed - // Right now it's fine b/c we limit the download from AWS to 32MB. We should recomment users to use parquet + // Right now it's fine b/c we limit the download from AWS to 32MB. We should recommend users to use parquet // for larger files let file_content_bytes = s3_object - .body - .collect() - .await - .map_err(|err| error::Error::InternalErr(err.to_string()))? - .into_bytes(); - let cursor = std::io::Cursor::new(file_content_bytes); + .take(length as usize) + .filter(|obj| future::ready(obj.is_ok())) + .map(|obj| obj.unwrap()) + .collect::>() + .await; + + let cursor = std::io::Cursor::new(file_content_bytes.concat()); let csv_df = CsvReader::new(cursor) .with_n_rows(Some(10)) // for now read only first 10 lines .with_separator(separator_final) .has_header(has_header.unwrap_or(true)) .finish() - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|err| { + tracing::error!("Error reading CSV file: {:?}", err); + Error::InternalErr(format!("Error reading CSV file: {}", err.to_string())) + })?; return Ok(format!("{:?}", csv_df).to_string()); } diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 7b90a32b60..4857f3fe80 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -49,6 +49,5 @@ regex.workspace = true git-version.workspace = true cron.workspace = true tracing-loki = { version = "^0", optional = true } -aws-sdk-s3.workspace = true -aws-config.workspace = true magic-crypt.workspace = true +object_store.workspace = true diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index 46ac255073..2c2b181458 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -1,6 +1,8 @@ -use aws_config::{BehaviorVersion, Region}; -use aws_sdk_s3::config::Credentials; +use crate::error; +use object_store::ObjectStore; +use object_store::{aws::AmazonS3Builder, ClientOptions}; use serde::{Deserialize, Serialize}; +use std::sync::Arc; #[derive(Serialize, Deserialize, Debug)] #[serde(tag = "type")] @@ -40,20 +42,20 @@ pub struct S3Object { } pub async fn get_etag_or_empty(s3_resource: &S3Resource, s3_object: S3Object) -> Option { - let s3_client = build_s3_client(s3_resource); + let s3_client = build_object_store_client(s3_resource); + if s3_client.is_err() { + return None; + } - 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 + let s3_key = object_store::path::Path::from(s3_object.s3); + + return s3_client + .unwrap() + .head(&s3_key) + .await .ok() - .map(|res| res.e_tag().map(|et| et.to_string())) - .flatten() + .map(|meta| meta.e_tag) + .flatten(); } pub fn render_endpoint(s3_resource: &S3Resource) -> String { @@ -73,26 +75,39 @@ pub fn render_endpoint(s3_resource: &S3Resource) -> String { } } -pub fn build_s3_client(s3_resource_ref: &S3Resource) -> aws_sdk_s3::Client { +pub fn build_object_store_client( + s3_resource_ref: &S3Resource, +) -> error::Result> { 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", - )); + + let mut store_builder = AmazonS3Builder::new() + .with_client_options(ClientOptions::new().with_timeout_disabled()) // TODO: make it configurable maybe + .with_region(s3_resource.region) + .with_bucket_name(s3_resource.bucket) + .with_endpoint(endpoint); + + if !s3_resource.use_ssl { + store_builder = store_builder.with_allow_http(true) } - if s3_resource.path_style { - s3_config_builder = s3_config_builder.force_path_style(true); + + if let Some(key) = s3_resource.access_key { + store_builder = store_builder.with_access_key_id(key); } - let s3_config = s3_config_builder.build(); - return aws_sdk_s3::Client::from_conf(s3_config); + if let Some(secret_key) = s3_resource.secret_key { + store_builder = store_builder.with_secret_access_key(secret_key); + } + if !s3_resource.path_style { + store_builder = store_builder.with_virtual_hosted_style_request(s3_resource.path_style); + } + + let store = store_builder.build().map_err(|err| { + tracing::error!("Error building object store client: {:?}", err); + error::Error::InternalErr(format!( + "Error building object store client: {}", + err.to_string() + )) + })?; + + return Ok(Arc::new(store)); } diff --git a/frontend/src/lib/components/S3FilePicker.svelte b/frontend/src/lib/components/S3FilePicker.svelte index d3f27ff14b..77fba7751a 100644 --- a/frontend/src/lib/components/S3FilePicker.svelte +++ b/frontend/src/lib/components/S3FilePicker.svelte @@ -25,7 +25,7 @@ let deletionModalOpen = false let fileDeletionInProgress = false - let fileListUnavailable = true + let fileListUnavailable: boolean | undefined = undefined let moveModalOpen = false let moveDestKey: string | undefined = undefined @@ -63,7 +63,6 @@ } > = {} let displayedFileKeys: string[] = [] - let paginationMarker: string | undefined = undefined let listDivHeight: number = 0 @@ -85,13 +84,41 @@ } | undefined = undefined + let listMarkers: string[] + let page = 0 + + const maxKeys = 1000 + + let count = 0 + + let filter = '' + + let timeout: NodeJS.Timeout | undefined = undefined + let firstLoad = true + $: filter != undefined && onFilterChange() + + function onFilterChange() { + if (!firstLoad) { + timeout && clearTimeout(timeout) + timeout = setTimeout(() => { + page = 0 + listMarkers = [] + loadFiles() + }, 500) + } else { + firstLoad = false + } + } + async function loadFiles() { fileListLoading = true let availableFiles = await HelpersService.listStoredFiles({ workspace: $workspaceStore!, - maxKeys: 1000, // fixed pages of 1000 files for now - marker: paginationMarker + maxKeys: maxKeys, // fixed pages of 1000 files for now + marker: page == 0 ? undefined : listMarkers[page - 1], + prefix: filter.trim() != '' ? filter : undefined }) + console.log(availableFiles?.windmill_large_files?.length) if ( availableFiles.restricted_access === null || availableFiles.restricted_access === undefined || @@ -134,10 +161,14 @@ } } } - displayedFileKeys = displayedFileKeys.sort() - if (availableFiles.next_marker !== undefined) { - paginationMarker = availableFiles.next_marker + if (listMarkers.length == page) { + count = availableFiles.windmill_large_files.length + const nextMarker = + availableFiles.windmill_large_files?.[availableFiles.windmill_large_files.length - 1]?.s3 + if (nextMarker) listMarkers.push(nextMarker) } + displayedFileKeys = displayedFileKeys.sort() + // before returning, un-collapse the folders containing the selected file (if any) if (selectedFileKey !== undefined && !emptyString(selectedFileKey.s3)) { let split_path = selectedFileKey.s3.split('/') @@ -268,25 +299,16 @@ } displayedFileKeys = [] allFilesByKey = {} - paginationMarker = undefined + count = 0 + page = 0 + filter = '' + listMarkers = [] fileMetadata = undefined filePreview = undefined reloadContent() drawer.openDrawer?.() } - export async function downloadS3File(fileKey: string | undefined) { - if (fileKey === undefined) { - return - } - const downloadUrl = await HelpersService.generateDownloadUrl({ - workspace: $workspaceStore!, - fileKey: fileKey - }) - console.log('download URL ', downloadUrl.download_url) - window.open(downloadUrl.download_url, '_blank') - } - async function reloadContent() { if (initialFileKey !== undefined) { initialFileKeyInternalCopy = { ...initialFileKey } @@ -406,7 +428,7 @@ {/if} {:else} - {#if fileListUnavailable} + {#if fileListUnavailable == true}

@@ -423,68 +445,93 @@ >

{/if} -
+
{#if !fileListUnavailable} -
+
+
+ +
{#if fileListLoading === false && displayedFileKeys.length === 0}
- No files in the workspace S3 bucket + No files in the workspace S3 bucket at that prefix
{:else} - -
- {@const file_info = allFilesByKey[displayedFileKeys[index]]} +
+
selectItem(index)} - class={`flex flex-row h-full font-semibold text-xs items-center justify-start ${ - selectedFileKey !== undefined && selectedFileKey.s3 === file_info.full_key - ? 'bg-surface-hover' - : '' - } `} + slot="item" + let:index + let:style + {style} + class="hover:bg-surface-hover border" > + {@const file_info = allFilesByKey[displayedFileKeys[index]]}
selectItem(index)} + class={`flex flex-row h-full font-semibold text-xs items-center justify-start ${ + selectedFileKey !== undefined && selectedFileKey.s3 === file_info.full_key + ? 'bg-surface-hover' + : '' + } `} > - {#if file_info.type === 'folder'} - {#if file_info.collapsed}{:else}{/if} -
- {file_info.display_name} -
- {:else} - -
- {file_info.display_name} -
- {/if} +
+ {#if file_info.type === 'folder'} + {#if file_info.collapsed}{:else}{/if} +
+ {file_info.display_name} +
+ {:else} + +
+ {file_info.display_name} +
+ {/if} +
-
+
+
+
+
{count} items on this page
+
Page {page + 1}
+ + {#if count == maxKeys} + + + {/if} +
+ {#if fileListLoading === true} +
+ Loading content
-
- {#if !emptyString(paginationMarker)} - - {/if} - {#if fileListLoading === true} -
- Loading content -
- {/if} -
- + {/if} {/if}
{/if} @@ -508,9 +555,8 @@ title="Download file from S3" variant="border" color="light" - on:click={() => { - downloadS3File(fileMetadata?.fileKey) - }} + href={`/api/w/${$workspaceStore}/job_helpers/download_s3_file?file_key=${fileMetadata?.fileKey}`} + download={fileMetadata?.fileKey.split('/').pop() ?? 'unnamed_download.file'} startIcon={{ icon: Download }} iconOnly={true} /> diff --git a/frontend/src/lib/components/common/fileDownload/FileDownload.svelte b/frontend/src/lib/components/common/fileDownload/FileDownload.svelte index ee2a0537d1..8afdcffbe1 100644 --- a/frontend/src/lib/components/common/fileDownload/FileDownload.svelte +++ b/frontend/src/lib/components/common/fileDownload/FileDownload.svelte @@ -1,34 +1,18 @@ - + diff --git a/frontend/src/lib/components/common/fileUpload/FileUpload.svelte b/frontend/src/lib/components/common/fileUpload/FileUpload.svelte index a944937eea..4b3f7aeb95 100644 --- a/frontend/src/lib/components/common/fileUpload/FileUpload.svelte +++ b/frontend/src/lib/components/common/fileUpload/FileUpload.svelte @@ -5,11 +5,11 @@ import Button from '$lib/components/common/button/Button.svelte' import { sendUserToast } from '$lib/toast' import { workspaceStore } from '$lib/stores' - import { HelpersService, type UploadFilePart } from '$lib/gen' + import { HelpersService } from '$lib/gen' import { writable, type Writable } from 'svelte/store' import { Ban, CheckCheck, FileWarning, Files, RefreshCcw, Trash } from 'lucide-svelte' import { twMerge } from 'tailwind-merge' - import { createEventDispatcher } from 'svelte' + import { createEventDispatcher, onDestroy } from 'svelte' import { emptyString } from '$lib/utils' export let acceptedFileTypes: string[] | undefined = ['*'] @@ -43,11 +43,11 @@ } } + let xhr: XMLHttpRequest | undefined = undefined async function uploadFileToS3(fileToUpload: File, fileToUploadKey: string) { if (fileToUpload === undefined || fileToUploadKey === undefined) { return } - let path: string | undefined = undefined let fileExtension: string | undefined = undefined if (randomFileKey) { @@ -63,7 +63,6 @@ })) ?? fileToUploadKey : fileToUploadKey } - const uploadData: FileUploadData = { name: fileToUpload.name, size: fileToUpload.size, @@ -72,97 +71,112 @@ path: path, file: fileToUpload } - $fileUploads = [...$fileUploads, uploadData] - let upload_id: string | undefined = undefined - let parts: UploadFilePart[] = [] + // // Use a custom TransformStream to track upload progress + // const progressTrackingStream = new TransformStream({ + // transform(chunk, controller) { + // controller.enqueue(chunk) + // bytesUploaded += chunk.byteLength + // console.log('upload progress:', bytesUploaded / totalBytes) + // uploadData.progress = (bytesUploaded / totalBytes) * 100 + // }, + // flush(controller) { + // console.log('completed stream') + // } + // }) - let reader = fileToUpload?.stream().getReader() - let { value: chunk, done: readerDone } = await reader.read() - if (chunk === undefined || readerDone) { - sendUserToast('Error reading file, no data read', true) + try { + // const response = await HelpersService.multipartFileUpload({ + // workspace: $workspaceStore!, + // fileKey: path, + // fileExtension: fileExtension, + // s3ResourcePath: customS3ResourcePath?.split(':')[1], + // requestBody: fileToUpload.stream().pipeThrough(progressTrackingStream, {}) + // }) + + const params = new URLSearchParams() + if (path) { + params.append('file_key', path) + } + if (customS3ResourcePath?.split(':')[1]) { + params.append('s3_resource_path', customS3ResourcePath?.split(':')[1]) + } + if (fileExtension) { + params.append('file_extension', fileExtension) + } + // let response = await fetch( + // `/api/w/${$workspaceStore}/job_helpers/multipart_upload_s3_file?${params.toString()}`, + // { + // method: 'POST', + // headers: { + // 'Content-Type': 'application/octet-stream' + // }, + // body: fileToUpload.stream().pipeThrough(progressTrackingStream, {}), + // duplex: 'half' + // } + // ) + + xhr = new XMLHttpRequest() + const response = (await new Promise((resolve, reject) => { + xhr?.upload.addEventListener('progress', (event) => { + if (event.lengthComputable) { + let progress = (event.loaded / event.total) * 100 + if (progress == 100) { + progress = 99 + } + console.log('upload progress:', progress) + uploadData.progress = progress + $fileUploads = $fileUploads + } + }) + xhr?.addEventListener('loadend', () => { + let response = xhr?.responseText + if (xhr?.readyState === 4 && xhr?.status === 200 && response) { + uploadData.progress = 100 + resolve(JSON.parse(response)) + } else { + if (response) { + reject('An error occurred while uploading the file, see server logs') + } else { + reject(response) + } + } + xhr = undefined + }) + xhr?.open( + 'POST', + `/api/w/${$workspaceStore}/job_helpers/upload_s3_file?${params.toString()}`, + true + ) + xhr?.setRequestHeader('Content-Type', 'application/octet-stream') + xhr?.send(fileToUpload) + })) as any + + uploadData.path = response.file_key + } catch (e) { + console.error(e) + sendUserToast(e, true) + $fileUploads = $fileUploads.map((fileUpload) => { + if (fileUpload.name === uploadData.name) { + fileUpload.errorMessage = e + return fileUpload + } + return fileUpload + }) return } + dispatch('addition', { path: uploadData.path }) + sendUserToast('File upload finished!') - let fileUploadProgress = 0 - while (true) { - const currentFileUpload = $fileUploads.find( - (fileUpload) => fileUpload.name === uploadData.name - )! - - if (currentFileUpload.cancelled) { - return + uploadData.progress = 100 + $fileUploads = $fileUploads.map((fileUpload) => { + if (fileUpload.name === uploadData.name) { + return uploadData } - - let { value: chunk_2, done: readerDone } = await reader.read() - if (!readerDone && chunk_2 !== undefined && chunk.length <= 5 * 1024 * 1024) { - // AWS enforces part to be bigger than 5MB, so we accumulate bytes until we reach that limit before triggering the request to the BE - chunk = new Uint8Array([...chunk, ...chunk_2]) - continue - } - - try { - let response = await HelpersService.multipartFileUpload({ - workspace: $workspaceStore!, - requestBody: { - file_key: path, - file_extension: fileExtension, - part_content: Array.from(chunk), - upload_id: upload_id, - parts: parts, - is_final: readerDone, - cancel_upload: currentFileUpload.cancelled ?? false, - s3_resource_path: - customS3ResourcePath !== undefined ? customS3ResourcePath.split(':')[1] : undefined, - file_expiration: undefined - } - }) - uploadData.path = response.file_key - path = response.file_key - upload_id = response.upload_id - parts = response.parts - - // update upload progress - fileUploadProgress += (chunk.length * 100) / fileToUpload.size - uploadData.progress = fileUploadProgress - $fileUploads = $fileUploads.map((fileUpload) => { - if (fileUpload.name === uploadData.name) { - return uploadData - } - return fileUpload - }) - - if (response.is_done) { - if (currentFileUpload.cancelled) { - sendUserToast('File upload cancelled!') - } else { - dispatch('addition', { path: path }) - sendUserToast('File upload finished!') - } - - break - } - if (chunk_2 === undefined) { - sendUserToast( - 'File upload is not finished, yet there is no more data to stream. This is unexpected', - true - ) - return - } - chunk = chunk_2 - } catch (e) { - sendUserToast(e, true) - $fileUploads = $fileUploads.map((fileUpload) => { - if (fileUpload.name === uploadData.name) { - fileUpload.errorMessage = e - return fileUpload - } - return fileUpload - }) - return - } - } + return fileUpload + }) + return } async function deleteFile(fileKey: string) { @@ -173,6 +187,13 @@ dispatch('deletion', { path: fileKey }) sendUserToast('File deleted!') } + + onDestroy(() => { + if (xhr) { + xhr?.abort + xhr = undefined + } + })
@@ -209,6 +230,11 @@ return } + if (xhr) { + xhr.abort() + xhr = undefined + } + $fileUploads = $fileUploads.filter( (_fileUpload) => _fileUpload.name !== fileUpload.name ) @@ -232,6 +258,11 @@ return } + if (xhr) { + xhr.abort() + xhr = undefined + } + $fileUploads = $fileUploads.filter( (_fileUpload) => _fileUpload.name !== fileUpload.name ) @@ -273,6 +304,10 @@ if (fileUpload.path) { deleteFile(fileUpload.path) } + if (xhr) { + xhr.abort() + xhr = undefined + } }} startIcon={{ icon: Trash diff --git a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte index 64206e197c..13c088b3cd 100644 --- a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte +++ b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte @@ -1,5 +1,5 @@