From ab897921879dddb671bd4f4774fff3ecb0a70bba Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 4 Apr 2024 22:06:55 +0200 Subject: [PATCH] nit global cache python improvements --- backend/Cargo.lock | 1 - backend/windmill-worker/Cargo.toml | 3 +-- backend/windmill-worker/src/global_cache.rs | 28 ++++++++++----------- 3 files changed, 15 insertions(+), 17 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 9212b6871e..295fe7714a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -9985,7 +9985,6 @@ dependencies = [ "bytes", "chrono", "const_format", - "crc", "deno_ast", "deno_console", "deno_core", diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index b6eb7cdf04..d0b080fe2e 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -14,7 +14,7 @@ prometheus = ["dep:prometheus", "windmill-common/prometheus"] enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:sha2", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"] benchmark = ["windmill-queue/benchmark"] flamegraph = [] -parquet = ["windmill-common/parquet", "dep:object_store", "dep:tar", "dep:crc"] +parquet = ["windmill-common/parquet", "dep:object_store", "dep:tar"] flow_testing = [] [dependencies] @@ -80,7 +80,6 @@ tokio-util = { workspace = true, optional = true } openidconnect = { workspace = true, optional = true} tar = { workspace = true, optional = true} object_store = { workspace = true, optional = true} -crc = { workspace = true, optional = true} [build-dependencies] deno_fetch.workspace = true diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index 86fcaa1967..88d8cfc288 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -1,5 +1,5 @@ #[cfg(all(feature = "enterprise", feature = "parquet"))] -use crate::{PIP_CACHE_DIR, ROOT_CACHE_DIR}; +use crate::PIP_CACHE_DIR; // #[cfg(feature = "enterprise")] // use rand::Rng; @@ -71,13 +71,9 @@ pub async fn build_tar_and_push( Ok(()) } -#[cfg(all(feature = "enterprise", feature = "parquet"))] -pub const X25: crc::Crc = crc::Crc::::new(&crc::CRC_16_IBM_SDLC); - #[cfg(all(feature = "enterprise", feature = "parquet"))] pub async fn pull_from_tar(client: Arc, folder: String) -> error::Result<()> { use object_store::path::Path; - use tokio::fs::metadata; let folder_name = folder.split("/").last().unwrap(); tracing::info!("Attempting to pull piptar {folder_name} from bucket"); @@ -85,9 +81,7 @@ pub async fn pull_from_tar(client: Arc, folder: String) -> erro let start = Instant::now(); let tar_path = format!("tar/pip/{folder_name}.tar"); - let object = client - .get(&Path::from(format!("tar/pip/{folder_name}.tar"))) - .await; + let object = client.get(&Path::from(tar_path.clone())).await; if let Err(e) = object { tracing::info!("Failed to pull tar from s3: {tar_path}. Error: {:?}", e); return Err(error::Error::ExecutionErr(format!( @@ -95,12 +89,18 @@ pub async fn pull_from_tar(client: Arc, folder: String) -> erro ))); } - let bytes = object.unwrap().bytes().await.unwrap(); - tracing::info!( - "{tar_path} checksum: {}, len: {}", - X25.checksum(&bytes), - bytes.len() - ); + let bytes = object.unwrap().bytes().await; + if bytes.is_err() { + tracing::info!( + "Failed to read tar from s3: {tar_path}. Error: {:?}", + bytes.err() + ); + return Err(error::Error::ExecutionErr(format!( + "Failed to read tar from s3: {tar_path}" + ))); + } + let bytes = bytes.unwrap(); + tracing::info!("{tar_path} len: {}", bytes.len()); if bytes.len() == 0 { tracing::info!(