From be624b1a9a39131ede5cf9a0b1676759140fa27a Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 5 Dec 2024 22:45:45 +0100 Subject: [PATCH 01/36] load all branchall even if branches > 20 --- frontend/src/lib/components/FlowStatusViewerInner.svelte | 8 ++++---- frontend/src/routes/view_graph/+page.svelte | 1 + 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index 53fa3b0af6..9f816d42c7 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -52,6 +52,7 @@ flowJobs: string[] flowJobsSuccess: (boolean | undefined)[] length: number + branchall?: boolean } | undefined = undefined @@ -466,9 +467,7 @@ let modId = flowJobIds?.moduleId ?? '' let common = { - iteration_from: - // $localDurationStatuses?.[modId]?.iteration_from ?? - Math.max(flowJobIds.flowJobs.length - 20, 0), + iteration_from: flowJobIds?.branchall ? 0 : Math.max(flowJobIds.flowJobs.length - 20, 0), iteration_total: $localDurationStatuses?.[modId]?.iteration_total ?? flowJobIds?.length } $localDurationStatuses[modId] = { @@ -1141,7 +1140,8 @@ moduleId: mod.id, flowJobs: mod.flow_jobs, flowJobsSuccess: mod.flow_jobs_success, - length: mod.iterator?.itered?.length ?? mod.flow_jobs.length + length: mod.iterator?.itered?.length ?? mod.flow_jobs.length, + branchall: job?.raw_flow?.modules?.[i]?.value?.type == 'branchall' } : undefined} on:jobsLoaded={(e) => { diff --git a/frontend/src/routes/view_graph/+page.svelte b/frontend/src/routes/view_graph/+page.svelte index 18b49846a8..a1804647ac 100644 --- a/frontend/src/routes/view_graph/+page.svelte +++ b/frontend/src/routes/view_graph/+page.svelte @@ -27,3 +27,4 @@ ) )}>Download +the 55 From 9ecc94a268d8540224a4111ced36e235407359b2 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 5 Dec 2024 23:52:34 +0100 Subject: [PATCH 02/36] optimize for loop rendering with more than 500 iterations --- frontend/src/lib/components/FlowStatusViewerInner.svelte | 9 ++++++++- .../components/flows/content/FlowModuleEarlyStop.svelte | 4 ++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index 9f816d42c7..35803d53e3 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -817,6 +817,8 @@ return rec(ids, undefined) } + + let subflowsSize = 500 {#if notAnonynmous} @@ -946,7 +948,12 @@
{#each flowJobIds?.flowJobs ?? [] as loopJobId, j (loopJobId)} - {#if render} + {#if render && j + subflowsSize + 1 == (flowJobIds?.flowJobs.length ?? 0)} + + {/if} + {#if render && j + subflowsSize + 1 > (flowJobIds?.flowJobs.length ?? 0)} +
Delay
{/if} {:else if delayType === 'exponential'} {#if flowModuleRetry?.exponential}
Attempts
- +
+ + +
Multiplier
delay = multiplier * base ^ (number of attempt) @@ -127,9 +149,9 @@ multiplier, random_factor } = flowModuleRetry?.exponential || {}} - {@const cArray = Array.from({ length: cAttempts || 0 }, () => cSeconds)} + {@const cArray = Array.from({ length: Math.min(cAttempts || 0, 100) }, () => cSeconds)} {@const eArray = Array.from( - { length: eAttempts || 0 }, + { length: Math.min(eAttempts || 0, 100) }, (_, i) => (multiplier || 0) * (eSeconds || 0) ** (i + cArray.length + 1) )} {@const array = [...cArray, ...eArray]} @@ -146,7 +168,7 @@ seconds){/if} - {#each array.slice(1) as delay, i} + {#each array.slice(1, 100) as delay, i} {@const index = i + 2} {index}: @@ -163,6 +185,12 @@ {/each} + {#if (cAttempts ?? 0) > 100 || (eAttempts ?? 0) > 100} + + ... + ... + + {/if} {:else}
No retries
From 45d4fc2de71540a7a9defec9877a5b32fe63d7a6 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 6 Dec 2024 01:24:36 +0100 Subject: [PATCH 04/36] bump deno to 2.1.2 and bun to 1.1.38 --- Dockerfile | 4 ++-- docker/DockerfileSlim | 2 +- docker/DockerfileSlimEe | 2 +- lsp/Dockerfile | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/Dockerfile b/Dockerfile index cff245d3e4..ef2f7a721e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -175,9 +175,9 @@ RUN /usr/local/bin/python3 -m pip install pip-tools COPY --from=builder /frontend/build /static_frontend COPY --from=builder /windmill/target/release/windmill ${APP}/windmill -COPY --from=denoland/deno:2.0.4 --chmod=755 /usr/bin/deno /usr/bin/deno +COPY --from=denoland/deno:2.1.2 --chmod=755 /usr/bin/deno /usr/bin/deno -COPY --from=oven/bun:1.1.34 /usr/local/bin/bun /usr/bin/bun +COPY --from=oven/bun:1.1.38 /usr/local/bin/bun /usr/bin/bun COPY --from=php:8.3.7-cli /usr/local/bin/php /usr/bin/php COPY --from=composer:2.7.6 /usr/bin/composer /usr/bin/composer diff --git a/docker/DockerfileSlim b/docker/DockerfileSlim index a1b5daca43..35b8b8ba6a 100644 --- a/docker/DockerfileSlim +++ b/docker/DockerfileSlim @@ -20,7 +20,7 @@ RUN /usr/local/bin/python3 -m pip install pip-tools # Install UV RUN curl --proto '=https' --tlsv1.2 -LsSf https://github.com/astral-sh/uv/releases/download/0.4.18/uv-installer.sh | sh && mv /root/.cargo/bin/uv /usr/local/bin/uv -COPY --from=oven/bun:1.1.34 /usr/local/bin/bun /usr/bin/bun +COPY --from=oven/bun:1.1.38 /usr/local/bin/bun /usr/bin/bun # add the docker client to call docker from a worker if enabled COPY --from=docker:dind /usr/local/bin/docker /usr/local/bin/ diff --git a/docker/DockerfileSlimEe b/docker/DockerfileSlimEe index 94cc160cae..44dd185d25 100644 --- a/docker/DockerfileSlimEe +++ b/docker/DockerfileSlimEe @@ -19,7 +19,7 @@ RUN /usr/local/bin/python3 -m pip install pip-tools # Install UV RUN curl --proto '=https' --tlsv1.2 -LsSf https://github.com/astral-sh/uv/releases/download/0.4.18/uv-installer.sh | sh && mv /root/.cargo/bin/uv /usr/local/bin/uv -COPY --from=oven/bun:1.1.34 /usr/local/bin/bun /usr/bin/bun +COPY --from=oven/bun:1.1.38 /usr/local/bin/bun /usr/bin/bun # add the docker client to call docker from a worker if enabled COPY --from=docker:dind /usr/local/bin/docker /usr/local/bin/ diff --git a/lsp/Dockerfile b/lsp/Dockerfile index 2d56305116..e805d48ec7 100644 --- a/lsp/Dockerfile +++ b/lsp/Dockerfile @@ -26,7 +26,7 @@ ENV GOBIN=/usr/local/go/bin RUN /usr/local/go/bin/go install -v golang.org/x/tools/gopls@latest RUN pip3 install tornado python-lsp-jsonrpc ruff-lsp -COPY --from=denoland/deno:2.0.2 --chmod=755 /usr/bin/deno /usr/bin/deno +COPY --from=denoland/deno:2.1.2 --chmod=755 /usr/bin/deno /usr/bin/deno COPY Pipfile . From f175158b9ffcc0d9c35938e61dfe3ce56d3b64c2 Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Fri, 6 Dec 2024 09:40:32 +0100 Subject: [PATCH 05/36] fix: handle flow & workspace renames for `flow_node` (#4861) --- ...5893edb73694601544fea7cf20f45dd25a88d.json | 27 +++++++++++++++++++ ...20241206075559_flow_node_unique_2.down.sql | 3 +++ .../20241206075559_flow_node_unique_2.up.sql | 3 +++ .../windmill-worker/src/worker_lockfiles.rs | 4 +-- 4 files changed, 34 insertions(+), 3 deletions(-) create mode 100644 backend/.sqlx/query-5af51d5bf7614274ade044120045893edb73694601544fea7cf20f45dd25a88d.json create mode 100644 backend/migrations/20241206075559_flow_node_unique_2.down.sql create mode 100644 backend/migrations/20241206075559_flow_node_unique_2.up.sql diff --git a/backend/.sqlx/query-5af51d5bf7614274ade044120045893edb73694601544fea7cf20f45dd25a88d.json b/backend/.sqlx/query-5af51d5bf7614274ade044120045893edb73694601544fea7cf20f45dd25a88d.json new file mode 100644 index 0000000000..f4021d8985 --- /dev/null +++ b/backend/.sqlx/query-5af51d5bf7614274ade044120045893edb73694601544fea7cf20f45dd25a88d.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO flow_node (path, workspace_id, hash_v2, lock, code, flow)\n VALUES ($1, $2, $3, $4, $5, $6)\n ON CONFLICT (path, workspace_id, hash_v2) DO UPDATE SET path = EXCLUDED.path -- trivial update to return the id\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Bpchar", + "Text", + "Text", + "Jsonb" + ] + }, + "nullable": [ + false + ] + }, + "hash": "5af51d5bf7614274ade044120045893edb73694601544fea7cf20f45dd25a88d" +} diff --git a/backend/migrations/20241206075559_flow_node_unique_2.down.sql b/backend/migrations/20241206075559_flow_node_unique_2.down.sql new file mode 100644 index 0000000000..b6e6db03e6 --- /dev/null +++ b/backend/migrations/20241206075559_flow_node_unique_2.down.sql @@ -0,0 +1,3 @@ +-- Add down migration script here +ALTER TABLE flow_node DROP CONSTRAINT IF EXISTS flow_node_unique_2; +ALTER TABLE flow_node ADD CONSTRAINT flow_node_hash_v2_key UNIQUE (hash_v2); diff --git a/backend/migrations/20241206075559_flow_node_unique_2.up.sql b/backend/migrations/20241206075559_flow_node_unique_2.up.sql new file mode 100644 index 0000000000..4e8383bd89 --- /dev/null +++ b/backend/migrations/20241206075559_flow_node_unique_2.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE flow_node ADD CONSTRAINT flow_node_unique_2 UNIQUE (path, workspace_id, hash_v2); +ALTER TABLE flow_node DROP CONSTRAINT IF EXISTS flow_node_hash_v2_key; diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 0ea7456ec3..ec739a88a2 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -1027,8 +1027,6 @@ async fn insert_flow_node<'c>( ) -> Result<(sqlx::Transaction<'c, sqlx::Postgres>, FlowNodeId)> { let hash = { let mut hasher = sha2::Sha256::new(); - hasher.update(path); - hasher.update(workspace_id); hasher.update(code.unwrap_or(&Default::default())); hasher.update(lock.unwrap_or(&Default::default())); hasher.update(flow.unwrap_or(&Default::default()).get()); @@ -1040,7 +1038,7 @@ async fn insert_flow_node<'c>( r#" INSERT INTO flow_node (path, workspace_id, hash_v2, lock, code, flow) VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (hash_v2) DO UPDATE SET path = EXCLUDED.path -- trivial update to return the id + ON CONFLICT (path, workspace_id, hash_v2) DO UPDATE SET path = EXCLUDED.path -- trivial update to return the id RETURNING id "#, path, From 3c4408e3dbf8a37c0b862805b1991af3d39ac054 Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Fri, 6 Dec 2024 12:26:36 +0100 Subject: [PATCH 06/36] feat(cache): refurbish fs backed cache (#4863) * feat(cache): refurbish fs backed cache * add `cached` exemple --- backend/windmill-common/src/cache.rs | 265 +++++++++++++++++++------ backend/windmill-common/src/flows.rs | 6 + backend/windmill-common/src/scripts.rs | 6 + 3 files changed, 213 insertions(+), 64 deletions(-) diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 9810e83be5..9a81c7dace 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -1,25 +1,149 @@ use crate::error; +use std::future::Future; +use std::hash::Hash; use std::path::{Path, PathBuf}; -use quick_cache::sync::Cache; +use quick_cache::Equivalent; +use serde::{Deserialize, Serialize}; use sqlx::PgExecutor; +pub use const_format::concatcp; +pub use lazy_static::lazy_static; +pub use quick_cache::sync::Cache; + /// Cache directory for windmill server/worker(s). pub const CACHE_DIR: &str = "/tmp/windmill/cache/"; +/// A file-system backed concurrent cache. +pub struct FsBackedCache { + cache: Cache, + root: &'static str, +} + +impl FsBackedCache { + /// Create a new file-system backed cache with `items_capacity` capacity. + /// The cache will be stored in the `root` directory. + pub fn new(root: &'static str, items_capacity: usize) -> Self { + Self { cache: Cache::new(items_capacity), root } + } + + /// Gets or inserts an item in the cache with key `key`. + pub async fn get_or_insert_async<'a, Q, F>(&'a self, key: &Q, with: F) -> error::Result + where + Q: Hash + Equivalent + ToOwned + Copy + Into, + F: Future>, + { + self.cache + .get_or_insert_async( + key, + fs::import_or_insert_with(self.root, (*key).into(), with), + ) + .await + } +} + +/// Like [`lazy_static`]`, but for file-system backed caches. +/// +/// # Example +/// ```rust +/// use windmill_common::make_static; +/// +/// make_static! { +/// /// String cache with a maximum capacity of 1000 items stored in the +/// /// "subdirectory" directory. +/// static ref CACHE: { u64 => String } in "subdirectory" <= 1000; +/// /// Another cache. +/// static ref ANOTHER_CACHE: { u64 => Vec } in "another" <= 100; +/// } +/// ``` +#[macro_export] +macro_rules! make_static { + { $( $(#[$attr:meta])* static ref $name:ident: { $Key:ty => $Val:ty } in $root:literal <= $cap:literal; )+ } => { + $crate::cache::lazy_static! { + $( + $(#[$attr])* + static ref $name: $crate::cache::FsBackedCache<$Key, $Val> = + $crate::cache::FsBackedCache::new( + $crate::cache::concatcp!($crate::cache::CACHE_DIR, $root), + $cap + ); + )+ + } + }; +} + +// re-export: +pub use make_static; + +/// Create an anonymous file-system backed cache for one-time use. +/// +/// # Example +/// ```rust +/// use windmill_common::anon; +/// let cache = anon!({ u64 => String } in "subdirectory" <= 1000); +/// ``` +#[macro_export] +macro_rules! anon { + ({ $Key:ty => $Val:ty } in $root:literal <= $cap:literal) => {{ + $crate::cache::make_static! { + static ref __ANON__: { $Key => $Val } in $root <= $cap; + } + + &__ANON__ + }}; +} + +// re-export: +pub use anon; + +pub mod future { + use super::*; + + /// Extension trait for futures that can be cached. + pub trait FutureCachedExt: + Future> + Sized + { + /// Get or insert the future result in the cache. + /// + /// # Example + /// ```rust + /// use windmill_common::cache::{self, future::FutureCachedExt}; + /// + /// async { + /// let result = std::future::ready(Ok(42)) + /// .cached(cache::anon!({ u64 => u64 } in "test" <= 1), &42) + /// .await; + /// + /// assert_eq!(result.unwrap(), 42); + /// }; + /// ``` + fn cached( + self, + cache: &FsBackedCache, + key: &Q, + ) -> impl Future> + where + Q: Hash + Equivalent + ToOwned + Copy + Into, + { + cache.get_or_insert_async(key, self) + } + } + + impl> + Sized> + FutureCachedExt for F + { + } +} + pub mod flow { use super::*; use crate::flows::{FlowNodeId, FlowValue}; - /// Cache directory for windmill server/worker(s) flow nodes. - pub const CACHE_DIR: &str = const_format::concatcp!(super::CACHE_DIR, "flow"); - - lazy_static::lazy_static! { + make_static! { /// Flow node cache. - /// FIXME: This should be a static but [`Cache`] does not have a const constructor. /// FIXME: Use `Arc` for cheap cloning. - static ref CACHE: Cache = Cache::new(1000); + static ref CACHE: { FlowNodeId => Val } in "flow" <= 1000; } /// Flow node cache value. @@ -75,38 +199,35 @@ pub mod flow { // so only one thread will be able to fetch the data from the database and write it to // the file system and cache, hence no race on the file system. CACHE - .get_or_insert_async( - &node, - fs::import_or_insert_with(CACHE_DIR, node.0 as u64, async { - sqlx::query!( - "SELECT \ + .get_or_insert_async(&node, async { + sqlx::query!( + "SELECT \ lock AS \"lock: String\", \ code AS \"code: String\", \ flow::text AS \"flow: Box\" \ FROM flow_node WHERE id = $1 LIMIT 1", - node.0, - ) - .fetch_one(e) - .await - .map_err(Into::into) - .and_then(|r| { - Ok(Val { - lock: r - .lock - .and_then(|x| if x.is_empty() { None } else { Some(x) }), - code: r.code, - flow: match r.flow { - None => None, - Some(flow) => serde_json::from_str(&flow).map_err(|err| { - error::Error::InternalErr(format!( - "Unable to parse flow value: {err:?}" - )) - })?, - }, - }) + node.0, + ) + .fetch_one(e) + .await + .map_err(Into::into) + .and_then(|r| { + Ok(Val { + lock: r + .lock + .and_then(|x| if x.is_empty() { None } else { Some(x) }), + code: r.code, + flow: match r.flow { + None => None, + Some(flow) => serde_json::from_str(&flow).map_err(|err| { + error::Error::InternalErr(format!( + "Unable to parse flow value: {err:?}" + )) + })?, + }, }) - }), - ) + }) + }) .await } @@ -164,14 +285,10 @@ pub mod script { use super::*; use crate::scripts::{ScriptHash, ScriptLang}; - /// Cache directory for windmill server/worker(s) scripts. - pub const CACHE_DIR: &str = const_format::concatcp!(super::CACHE_DIR, "script"); - - lazy_static::lazy_static! { + make_static! { /// Scripts cache. - /// FIXME: This should be a static but [`Cache`] does not have a const constructor. /// FIXME: Use `Arc` for cheap cloning. - static ref CACHE: Cache = Cache::new(1000); + static ref CACHE: { ScriptHash => Val } in "script" <= 1000; } /// Script cache value. @@ -197,34 +314,31 @@ pub mod script { // so only one thread will be able to fetch the data from the database and write it to // the file system and cache, hence no race on the file system. CACHE - .get_or_insert_async( - &hash, - fs::import_or_insert_with(CACHE_DIR, hash.0 as u64, async { - sqlx::query!( - "SELECT \ + .get_or_insert_async(&hash, async { + sqlx::query!( + "SELECT \ lock AS \"lock: String\", \ content AS \"code!: String\", language AS \"language: Option\", \ envs AS \"envs: Vec\", \ codebase AS \"codebase: String\" \ FROM script WHERE hash = $1 AND workspace_id = $2 LIMIT 1", - hash.0, - workspace_id, - ) - .fetch_one(e) - .await - .map_err(Into::into) - .map(|r| Val { - lock: r - .lock - .and_then(|x| if x.is_empty() { None } else { Some(x) }), - code: r.code, - language: r.language, - envs: r.envs, - codebase: r.codebase, - }) - }), - ) + hash.0, + workspace_id, + ) + .fetch_one(e) + .await + .map_err(Into::into) + .map(|r| Val { + lock: r + .lock + .and_then(|x| if x.is_empty() { None } else { Some(x) }), + code: r.code, + language: r.language, + envs: r.envs, + codebase: r.codebase, + }) + }) .await } @@ -283,8 +397,6 @@ pub mod script { mod fs { use super::*; - use std::future::Future; - use std::fs::{self, OpenOptions}; use std::io::{Read, Write}; @@ -363,4 +475,29 @@ mod fs { } Ok(data) } + + // Auto-implement `Bundle` for all `serde` serializable types. + + impl Item for () { + fn path(&self, root: &Path) -> PathBuf { + root.join("self.json") + } + } + + impl Deserialize<'de> + Serialize + Default> Bundle for T { + type Item = (); + + fn items() -> &'static [Self::Item] { + &[()] + } + + fn import(&mut self, _: Self::Item, data: Vec) -> error::Result<()> { + *self = serde_json::from_slice(&data)?; + Ok(()) + } + + fn export(&self, _: Self::Item) -> error::Result>> { + Ok(Some(serde_json::to_vec(self)?)) + } + } } diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index cd67f741a2..29edd4db30 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -407,6 +407,12 @@ pub enum InputTransform { #[serde(transparent)] pub struct FlowNodeId(pub i64); +impl Into for FlowNodeId { + fn into(self) -> u64 { + self.0 as u64 + } +} + #[derive(Serialize, Deserialize, Debug, Clone)] pub struct Branch { #[serde(skip_serializing_if = "Option::is_none")] diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 1a11b5a701..9d51bbb830 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -75,6 +75,12 @@ impl ScriptLang { #[sqlx(transparent)] pub struct ScriptHash(pub i64); +impl Into for ScriptHash { + fn into(self) -> u64 { + self.0 as u64 + } +} + #[derive(PartialEq, sqlx::Type)] #[sqlx(transparent, no_pg_array)] pub struct ScriptHashes(pub Vec); From 2bc4934c4f38ca2631eb99140d89a0b3d3b3a2cf Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Fri, 6 Dec 2024 13:37:45 +0100 Subject: [PATCH 07/36] feat: add db storage for app inline scripts (#4837) --- ...9177a87e1accd192402e21db5ae09c3498ab0.json | 3 +- ...54799f5e8122adaf35465cb16c3dc795bdc3b.json | 26 ++ ...242cf3a6287e7b4548c7f01ad888230c27013.json | 22 ++ ...98a03f751b246c40daf056fced0fd91f6dd73.json | 3 +- ...ffe0ba043565790865c6a6f52fab6ce340d2c.json | 23 ++ ...31899161e80c239d56eb657d73bbf4272939b.json | 28 ++ ...914fca6fc0a8787e4f8d520766b1c89f23602.json | 15 + ...b6b493e53583017c18e2ab44f44125c52d548.json | 3 +- ...66b142fa628b8983b966357deb3d5e0a1df3c.json | 28 ++ .../20241202095902_app_script.down.sql | 3 + .../20241202095902_app_script.up.sql | 26 ++ backend/windmill-api/openapi.yaml | 21 ++ backend/windmill-api/src/apps.rs | 318 ++++++++++-------- backend/windmill-common/src/apps.rs | 13 +- backend/windmill-common/src/cache.rs | 90 +++++ backend/windmill-common/src/jobs.rs | 8 + backend/windmill-queue/src/jobs.rs | 23 +- backend/windmill-worker/src/worker.rs | 15 + .../windmill-worker/src/worker_lockfiles.rs | 101 +++++- .../helpers/RunnableComponent.svelte | 11 +- frontend/src/lib/components/apps/types.ts | 2 + .../(logged)/apps/get/[...path]/+page.svelte | 2 +- 22 files changed, 632 insertions(+), 152 deletions(-) create mode 100644 backend/.sqlx/query-0c6c80746733be8f561ab0b631854799f5e8122adaf35465cb16c3dc795bdc3b.json create mode 100644 backend/.sqlx/query-1bae415f9440cc1334f24ce3009242cf3a6287e7b4548c7f01ad888230c27013.json create mode 100644 backend/.sqlx/query-5d7081a9ba0d702f63ed9d44be5ffe0ba043565790865c6a6f52fab6ce340d2c.json create mode 100644 backend/.sqlx/query-ea9bbb972217bab4d7e8f4c08e331899161e80c239d56eb657d73bbf4272939b.json create mode 100644 backend/.sqlx/query-ee16199b4af456198fae062e948914fca6fc0a8787e4f8d520766b1c89f23602.json create mode 100644 backend/.sqlx/query-ffa86babfcab107caffb8dda31a66b142fa628b8983b966357deb3d5e0a1df3c.json create mode 100644 backend/migrations/20241202095902_app_script.down.sql create mode 100644 backend/migrations/20241202095902_app_script.up.sql diff --git a/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json b/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json index 8fdca62eca..1b1c6b504b 100644 --- a/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json +++ b/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json @@ -44,7 +44,8 @@ "deploymentcallback", "singlescriptflow", "flowscript", - "flownode" + "flownode", + "appscript" ] } } diff --git a/backend/.sqlx/query-0c6c80746733be8f561ab0b631854799f5e8122adaf35465cb16c3dc795bdc3b.json b/backend/.sqlx/query-0c6c80746733be8f561ab0b631854799f5e8122adaf35465cb16c3dc795bdc3b.json new file mode 100644 index 0000000000..efe58b7a16 --- /dev/null +++ b/backend/.sqlx/query-0c6c80746733be8f561ab0b631854799f5e8122adaf35465cb16c3dc795bdc3b.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO app_script (app, hash, lock, code, code_sha256)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (hash) DO UPDATE SET app = EXCLUDED.app -- trivial update to return the id\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8", + "Bpchar", + "Text", + "Text", + "Bpchar" + ] + }, + "nullable": [ + false + ] + }, + "hash": "0c6c80746733be8f561ab0b631854799f5e8122adaf35465cb16c3dc795bdc3b" +} diff --git a/backend/.sqlx/query-1bae415f9440cc1334f24ce3009242cf3a6287e7b4548c7f01ad888230c27013.json b/backend/.sqlx/query-1bae415f9440cc1334f24ce3009242cf3a6287e7b4548c7f01ad888230c27013.json new file mode 100644 index 0000000000..5c43791411 --- /dev/null +++ b/backend/.sqlx/query-1bae415f9440cc1334f24ce3009242cf3a6287e7b4548c7f01ad888230c27013.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT format('rawscript/%s', code_sha256) as \"path!: String\"\n FROM app_script WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path!: String", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "1bae415f9440cc1334f24ce3009242cf3a6287e7b4548c7f01ad888230c27013" +} diff --git a/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json b/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json index 20d76cbed4..2f80c0c818 100644 --- a/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json +++ b/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json @@ -34,7 +34,8 @@ "deploymentcallback", "singlescriptflow", "flowscript", - "flownode" + "flownode", + "appscript" ] } } diff --git a/backend/.sqlx/query-5d7081a9ba0d702f63ed9d44be5ffe0ba043565790865c6a6f52fab6ce340d2c.json b/backend/.sqlx/query-5d7081a9ba0d702f63ed9d44be5ffe0ba043565790865c6a6f52fab6ce340d2c.json new file mode 100644 index 0000000000..c9cd8219ee --- /dev/null +++ b/backend/.sqlx/query-5d7081a9ba0d702f63ed9d44be5ffe0ba043565790865c6a6f52fab6ce340d2c.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT policy as \"policy: sqlx::types::Json>\"\n FROM app WHERE app.path = $1 AND app.workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "policy: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "5d7081a9ba0d702f63ed9d44be5ffe0ba043565790865c6a6f52fab6ce340d2c" +} diff --git a/backend/.sqlx/query-ea9bbb972217bab4d7e8f4c08e331899161e80c239d56eb657d73bbf4272939b.json b/backend/.sqlx/query-ea9bbb972217bab4d7e8f4c08e331899161e80c239d56eb657d73bbf4272939b.json new file mode 100644 index 0000000000..450683521d --- /dev/null +++ b/backend/.sqlx/query-ea9bbb972217bab4d7e8f4c08e331899161e80c239d56eb657d73bbf4272939b.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT app_id, value FROM app_version WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "app_id", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "value", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "ea9bbb972217bab4d7e8f4c08e331899161e80c239d56eb657d73bbf4272939b" +} diff --git a/backend/.sqlx/query-ee16199b4af456198fae062e948914fca6fc0a8787e4f8d520766b1c89f23602.json b/backend/.sqlx/query-ee16199b4af456198fae062e948914fca6fc0a8787e4f8d520766b1c89f23602.json new file mode 100644 index 0000000000..0d76f9a721 --- /dev/null +++ b/backend/.sqlx/query-ee16199b4af456198fae062e948914fca6fc0a8787e4f8d520766b1c89f23602.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO app_version_lite (id, value) VALUES ($1, $2)\n ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "ee16199b4af456198fae062e948914fca6fc0a8787e4f8d520766b1c89f23602" +} diff --git a/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json b/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json index 4092b60b71..e48e3b041d 100644 --- a/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json +++ b/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json @@ -48,7 +48,8 @@ "deploymentcallback", "singlescriptflow", "flowscript", - "flownode" + "flownode", + "appscript" ] } } diff --git a/backend/.sqlx/query-ffa86babfcab107caffb8dda31a66b142fa628b8983b966357deb3d5e0a1df3c.json b/backend/.sqlx/query-ffa86babfcab107caffb8dda31a66b142fa628b8983b966357deb3d5e0a1df3c.json new file mode 100644 index 0000000000..49fe5cf6b8 --- /dev/null +++ b/backend/.sqlx/query-ffa86babfcab107caffb8dda31a66b142fa628b8983b966357deb3d5e0a1df3c.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT lock, code FROM app_script WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "lock", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "code", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true, + false + ] + }, + "hash": "ffa86babfcab107caffb8dda31a66b142fa628b8983b966357deb3d5e0a1df3c" +} diff --git a/backend/migrations/20241202095902_app_script.down.sql b/backend/migrations/20241202095902_app_script.down.sql new file mode 100644 index 0000000000..c493d82176 --- /dev/null +++ b/backend/migrations/20241202095902_app_script.down.sql @@ -0,0 +1,3 @@ +-- Add down migration script here +DROP TABLE IF EXISTS app_version_lite; +DROP TABLE IF EXISTS app_script; diff --git a/backend/migrations/20241202095902_app_script.up.sql b/backend/migrations/20241202095902_app_script.up.sql new file mode 100644 index 0000000000..5c3e6a2e0b --- /dev/null +++ b/backend/migrations/20241202095902_app_script.up.sql @@ -0,0 +1,26 @@ +-- Add up migration script here +ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'appscript'; + +-- Same as `app_version` but with a "lite" value (w/ `inlineScript.{code,lock}`). +CREATE TABLE app_version_lite ( + id BIGSERIAL PRIMARY KEY, + value JSONB, + FOREIGN KEY (id) REFERENCES app_version (id) ON DELETE CASCADE +); + +GRANT ALL ON app_version_lite TO windmill_user; +GRANT ALL ON app_version_lite TO windmill_admin; + +-- App `inlineScript`. +CREATE TABLE app_script ( + id BIGSERIAL PRIMARY KEY, + app BIGSERIAL NOT NULL, + hash CHAR(64) NOT NULL UNIQUE, -- sha256 of `app`, `lock`, `code`. + lock TEXT, + code TEXT NOT NULL, + code_sha256 CHAR(64) NOT NULL, -- used to retrieve the policy. + FOREIGN KEY (app) REFERENCES app (id) ON DELETE CASCADE +); + +GRANT ALL ON app_script TO windmill_user; +GRANT ALL ON app_script TO windmill_admin; diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 95c06d5aae..ccab668051 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5467,6 +5467,23 @@ paths: schema: $ref: "#/components/schemas/AppWithLastVersion" + /w/{workspace}/apps/get/lite/{path}: + get: + summary: get app lite by path + operationId: getAppLiteByPath + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/ScriptPath" + responses: + "200": + description: app lite details + content: + application/json: + schema: + $ref: "#/components/schemas/AppWithLastVersion" + /w/{workspace}/apps/get/draft/{path}: get: summary: get app by path with draft @@ -5791,6 +5808,8 @@ paths: #flow: flow/ path: type: string + version: + type: integer args: {} raw_code: type: object @@ -5808,6 +5827,8 @@ paths: required: - content - language + id: + type: integer force_viewer_static_fields: type: object force_viewer_one_of_fields: diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index e1a9421252..af167d9a4a 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::{collections::HashMap, sync::Arc}; /* * Author: Ruben Fiszel @@ -31,6 +31,7 @@ use axum::{ routing::{delete, get, post}, Router, }; +use futures::future::{FutureExt, TryFutureExt}; use hyper::StatusCode; #[cfg(feature = "parquet")] use itertools::Itertools; @@ -50,7 +51,8 @@ use windmill_audit::ActionKind; #[cfg(feature = "parquet")] use windmill_common::s3_helpers::build_object_store_client; use windmill_common::{ - apps::ListAppQuery, + apps::{AppScriptId, ListAppQuery}, + cache::{self, future::FutureCachedExt}, db::UserDB, error::{to_anyhow, Error, JsonResult, Result}, jobs::{get_payload_tag_from_prefixed_path, JobPayload, RawCode}, @@ -71,6 +73,7 @@ pub fn workspaced_service() -> Router { .route("/list", get(list_apps)) .route("/list_search", get(list_search_apps)) .route("/get/p/*path", get(get_app)) + .route("/get/lite/*path", get(get_app_lite)) .route("/get/draft/*path", get(get_app_w_draft)) .route("/secret_of/*path", get(get_secret_id)) .route("/get/v/*id", get(get_app_by_id)) @@ -191,15 +194,16 @@ pub type StaticFields = HashMap>; pub type OneOfFields = HashMap>>; pub type AllowUserResources = Vec; -#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)] #[serde(rename_all = "lowercase")] pub enum ExecutionMode { + #[default] Anonymous, Publisher, Viewer, } -#[derive(Serialize, Deserialize, Debug, Clone)] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] pub struct PolicyTriggerableInputs { static_inputs: StaticFields, one_of_inputs: OneOfFields, @@ -215,7 +219,7 @@ pub struct S3Input { file_key_regex: String, } -#[derive(Serialize, Deserialize, Debug, Clone)] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] pub struct Policy { pub on_behalf_of: Option, pub on_behalf_of_email: Option, @@ -410,6 +414,33 @@ async fn get_app( Ok(Json(app)) } +async fn get_app_lite( + authed: ApiAuthed, + Extension(user_db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, +) -> JsonResult { + let path = path.to_path(); + let mut tx = user_db.begin(&authed).await?; + + let app_o = sqlx::query_as::<_, AppWithLastVersion>( + "SELECT app.id, app.path, app.summary, app.versions, app.policy, + app.extra_perms, coalesce(app_version_lite.value::json, app_version.value) as value, + app_version.created_at, app_version.created_by, NULL as starred + FROM app, app_version + LEFT JOIN app_version_lite ON app_version_lite.id = app_version.id + WHERE app.path = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]", + ) + .bind(path.to_owned()) + .bind(&w_id) + .fetch_optional(&mut *tx) + .await?; + + tx.commit().await?; + + let app = not_found_if_none(app_o, "App", path)?; + Ok(Json(app)) +} + async fn get_app_w_draft( authed: ApiAuthed, Extension(user_db): Extension, @@ -583,8 +614,9 @@ async fn get_public_app_by_secret( let app_o = sqlx::query_as::<_, AppWithLastVersion>( "SELECT app.id, app.path, app.summary, app.versions, app.policy, - null as extra_perms, app_version.value, + null as extra_perms, coalesce(app_version_lite.value::json, app_version.value::json) as value, app_version.created_at, app_version.created_by from app, app_version + LEFT JOIN app_version_lite ON app_version_lite.id = app_version.id WHERE app.id = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]") .bind(&id) .bind(&w_id) @@ -1144,6 +1176,10 @@ async fn update_app( #[derive(Debug, Deserialize, Clone)] pub struct ExecuteApp { + /// The app version to execute. Fallback to `path` if not provided. + pub version: Option, + /// The app script id (from the `app_script` table) to execute. + pub id: Option, pub args: HashMap>, // - script: script/ // - flow: flow/ @@ -1206,6 +1242,22 @@ async fn get_on_behalf_details_from_policy_and_authed( Ok((username, permissioned_as, email)) } +/// Convert the triggerables from the old format to the new format. +fn empty_triggerables(mut policy: Policy) -> Policy { + use std::mem::take; + if let Some(triggerables) = take(&mut policy.triggerables) { + let mut triggerables_v2 = take(&mut policy.triggerables_v2).unwrap_or_default(); + for (k, static_inputs) in triggerables.into_iter() { + triggerables_v2.insert( + k, + PolicyTriggerableInputs { static_inputs, ..Default::default() }, + ); + } + policy.triggerables_v2 = Some(triggerables_v2); + } + policy +} + async fn execute_component( OptAuthed(opt_authed): OptAuthed, Extension(db): Extension, @@ -1228,99 +1280,136 @@ async fn execute_component( }; let path = path.to_path(); + let (arc_policy, policy): (Arc, Policy); + let policy_triggerables_default = Default::default(); - let policy = match payload.clone() { + // Two cases here: + // 1. The component is executed from the editor (i.e. in "preview" mode), then: + // - The policy is set to default (in `Viewer` execution mode). + // - The policy triggerables are built by the frontend and retrieved from the request + // payload. + // - In case of inline script, the `RawCode` from the request is pushed as is to the + // job queue. + // 2. Otherwise (i.e. "run" mode): + // - The policy and triggerables are fetched from the database. + // - In case of inline script, if an entry exists in the `app_script` table, push + // an `AppScript` job payload, as in (.1) otherwise. + let (policy, policy_triggerables) = match payload { + // 1. "preview" mode. ExecuteApp { - force_viewer_static_fields: Some(static_fields), - force_viewer_one_of_fields: Some(one_of_fields), + force_viewer_static_fields: Some(static_inputs), + force_viewer_one_of_fields: Some(one_of_inputs), force_viewer_allow_user_resources: Some(allow_user_resources), .. - } => { - let mut hm = HashMap::new(); - - if let Some(path) = payload.path.clone() { - hm.insert( - format!("{}:{path}", payload.component), - PolicyTriggerableInputs { - static_inputs: static_fields, - one_of_inputs: one_of_fields, - allow_user_resources, - }, - ); - } else { - hm.insert( - format!( - "{}:{}", - payload.component, - digest(payload.raw_code.clone().unwrap().content.as_str()) - ), - PolicyTriggerableInputs { - static_inputs: static_fields, - one_of_inputs: one_of_fields, - allow_user_resources, - }, - ); - } - Policy { + } => ( + &Policy { execution_mode: ExecutionMode::Viewer, - triggerables: None, - triggerables_v2: Some(hm), - on_behalf_of: None, - on_behalf_of_email: None, - s3_inputs: None, - } - } + ..Default::default() + }, + &PolicyTriggerableInputs { + static_inputs, + one_of_inputs, + allow_user_resources, + }, + ), + // 2. "run" mode. _ => { - let policy_o = sqlx::query_scalar!( - "SELECT policy from app WHERE path = $1 AND workspace_id = $2", + // Policy is fetched from the database on app `path` and `workspace_id`. + let policy_fut = sqlx::query_scalar!( + "SELECT policy as \"policy: sqlx::types::Json>\" + FROM app WHERE app.path = $1 AND app.workspace_id = $2 LIMIT 1", path, - &w_id + &w_id, ) .fetch_optional(&db) - .await?; + .map_err(Into::::into) + .map(|policy_o| Result::Ok(not_found_if_none(policy_o?, "App", path)?)) + .map(|policy| Result::Ok(serde_json::from_str(policy?.get())?)) + .map_ok(empty_triggerables); - let policy = not_found_if_none(policy_o, "App", path)?; + // 1. The app `version` is provided: cache the fetched policy. + // 2. Otherwise, always fetch the policy from the database. + let policy = if let Some(id) = payload.version { + let cache = cache::anon!({ u64 => Arc } in "policy" <= 1000); + arc_policy = policy_fut + .map_ok(Arc::new) + .cached(cache, &(id as u64)) + .await?; + &*arc_policy + } else { + policy = policy_fut.await?; + &policy + }; - serde_json::from_value::(policy).map_err(to_anyhow)? + // Compute the path for the triggerables map: + // - flow: `flow/` + // - script: `script/` + // - inline script: `rawscript/` + let path = match &payload { + // flow or script: just use the `payload.path`. + ExecuteApp { path: Some(path), .. } => path, + // inline script: without entry in the `app_script` table. + ExecuteApp { raw_code: Some(raw_code), id: None, .. } => &digest(&raw_code.content), + // inline script: with an entry in the `app_script` table. + ExecuteApp { raw_code: Some(_), id: Some(id), .. } => { + let cache = cache::anon!({ u64 => Arc } in "appscriptpath" <= 10000); + // `id` is unique, cache the result. + &*sqlx::query_scalar!( + "SELECT format('rawscript/%s', code_sha256) as \"path!: String\" + FROM app_script WHERE id = $1 LIMIT 1", + id + ) + .fetch_one(&db) + .map_err(Into::::into) + .map_ok(Arc::new) + .cached(cache, &(*id as u64)) + .await? + } + _ => unreachable!(), + }; + + // Retrieve the triggerables from the policy on `path` or `:`. + let triggerables_v2 = policy + .triggerables_v2 + .as_ref() + .ok_or_else(|| Error::BadRequest(format!("Policy is missing triggerables")))?; + let policy_triggerables = triggerables_v2 + .get(path) // start with `path` in case we can avoid the next` format!`. + .or_else(|| triggerables_v2.get(&format!("{}:{}", payload.component, &path))) + .or(match policy.execution_mode { + ExecutionMode::Viewer => Some(&policy_triggerables_default), + _ => None, + }) + .ok_or_else(|| Error::BadRequest(format!("Path {path} forbidden by policy")))?; + + (policy, policy_triggerables) } }; let (username, permissioned_as, email) = get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?; - let (job_payload, (args, job_id), tag) = match payload { - ExecuteApp { args, component, raw_code: Some(raw_code), path: None, .. } => { - let content = &raw_code.content; - let payload = JobPayload::Code(raw_code.clone()); - let path = digest(content); - let args = build_args( - policy, - &component, - path, - args, - opt_authed.as_ref(), - &user_db, - &db, - &w_id, - ) - .await?; - (payload, args, None) - } - ExecuteApp { args, component, raw_code: None, path: Some(path), .. } => { - let (payload, tag) = get_payload_tag_from_prefixed_path(&path, &db, &w_id).await?; - let args = build_args( - policy, - &component, - path.to_string(), - args, - opt_authed.as_ref(), - &user_db, - &db, - &w_id, - ) - .await?; - (payload, args, tag) - } + let (args, job_id) = build_args( + policy, + policy_triggerables, + payload.args, + opt_authed.as_ref(), + &user_db, + &db, + &w_id, + ) + .await?; + + let (job_payload, tag) = match (payload.path, payload.raw_code, payload.id) { + // flow or script: + (Some(path), None, None) => get_payload_tag_from_prefixed_path(&path, &db, &w_id).await?, + // inline script: in "preview" mode or without entry in the `app_script` table. + (None, Some(raw_code), None) => (JobPayload::Code(raw_code), None), + // inline script: in "run" mode and with an entry in the `app_script` table. + (None, Some(RawCode { language, path, cache_ttl, .. }), Some(id)) => ( + JobPayload::AppScript { id: AppScriptId(id), cache_ttl, language, path }, + None, + ), _ => unreachable!(), }; let tx = windmill_queue::PushIsolationLevel::IsolatedRoot(db.clone()); @@ -1655,9 +1744,12 @@ async fn exists_app( } async fn build_args( - policy: Policy, - component: &str, - path: String, + policy: &Policy, + PolicyTriggerableInputs { + static_inputs, + one_of_inputs, + allow_user_resources, + }: &PolicyTriggerableInputs, mut args: HashMap>, authed: Option<&ApiAuthed>, user_db: &UserDB, @@ -1665,54 +1757,6 @@ async fn build_args( w_id: &str, ) -> Result<(PushArgsOwned, Option)> { let mut job_id: Option = None; - let key = format!("{}:{}", component, &path); - let (static_inputs, one_of_inputs, allow_user_resources) = match policy { - Policy { triggerables_v2: Some(t), .. } => { - let PolicyTriggerableInputs { static_inputs, one_of_inputs, allow_user_resources } = t - .get(&key) - .or_else(|| t.get(&path)) - .map(|x| x.clone()) - .or_else(|| { - if matches!(policy.execution_mode, ExecutionMode::Viewer) { - Some(PolicyTriggerableInputs { - static_inputs: HashMap::new(), - one_of_inputs: HashMap::new(), - allow_user_resources: Vec::new(), - }) - } else { - None - } - }) - .ok_or_else(|| { - Error::BadRequest(format!("path {} is not allowed in the app policy", path)) - })?; - - (static_inputs, one_of_inputs, allow_user_resources) - } - Policy { triggerables: Some(t), .. } => { - let static_inputs = t - .get(&key) - .or_else(|| t.get(&path)) - .map(|x| x.clone()) - .or_else(|| { - if matches!(policy.execution_mode, ExecutionMode::Viewer) { - Some(HashMap::new()) - } else { - None - } - }) - .ok_or_else(|| { - Error::BadRequest(format!("path {} is not allowed in the app policy", path)) - })?; - - (static_inputs, HashMap::new(), Vec::new()) - } - _ => Err(Error::BadRequest(format!( - "Policy is missing triggerables for {}", - key - )))?, - }; - let mut safe_args = HashMap::>::new(); // tracing::error!("{:?}", allow_user_resources); @@ -1761,16 +1805,16 @@ async fn build_args( } for (k, v) in one_of_inputs { - if safe_args.contains_key(&k) { + if safe_args.contains_key(k) { continue; } - if let Some(arg_val) = args.get(&k) { + if let Some(arg_val) = args.get(k) { let arg_str = arg_val.get(); let options_str_vec = v.iter().map(|x| x.get()).collect::>(); if options_str_vec.contains(&arg_str) { safe_args.insert(k.to_string(), arg_val.clone()); - args.remove(&k); + args.remove(k); continue; } @@ -1781,7 +1825,7 @@ async fn build_args( .all(|x| options_str_vec.contains(&x.get())) { safe_args.insert(k.to_string(), arg_val.clone()); - args.remove(&k); + args.remove(k); continue; } } diff --git a/backend/windmill-common/src/apps.rs b/backend/windmill-common/src/apps.rs index acbf720f57..2948fd7cb7 100644 --- a/backend/windmill-common/src/apps.rs +++ b/backend/windmill-common/src/apps.rs @@ -6,7 +6,18 @@ * LICENSE-AGPL for a copy of the license. */ -use serde::Deserialize; +use serde::{Deserialize, Serialize}; + +/// Id in the `app_script` table. +#[derive(Serialize, Deserialize, Debug, Copy, Clone, Hash, Eq, PartialEq)] +#[serde(transparent)] +pub struct AppScriptId(pub i64); + +impl Into for AppScriptId { + fn into(self) -> u64 { + self.0 as u64 + } +} #[derive(Deserialize)] pub struct ListAppQuery { diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 9a81c7dace..45fdd0bd82 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -394,6 +394,96 @@ pub mod script { } } +pub mod app { + use super::*; + use crate::apps::AppScriptId; + + make_static! { + /// App scripts cache. + /// FIXME: Use `Arc` for cheap cloning. + static ref CACHE: { AppScriptId => Val } in "app" <= 1000; + } + + /// App app script cache value. + #[derive(Debug, Clone, Default)] + pub struct Val { + pub lock: Option, + pub code: String, + } + + /// Fetch the app script referenced by `id` from the cache. + /// If not present, import from the file-system cache or fetch it from the database and write + /// it to the file system and cache. + /// This should be preferred over fetching the database directly. + pub async fn fetch_script( + e: impl PgExecutor<'_>, + id: AppScriptId, + ) -> error::Result<(Option, String)> { + // If not present, `get_or_insert_async` will lock the key until the future completes, + // so only one thread will be able to fetch the data from the database and write it to + // the file system and cache, hence no race on the file system. + CACHE + .get_or_insert_async(&id, async { + sqlx::query!( + "SELECT lock, code FROM app_script WHERE id = $1 LIMIT 1", + id.0, + ) + .fetch_one(e) + .await + .map_err(Into::into) + .map(|r| Val { + lock: r + .lock + .and_then(|x| if x.is_empty() { None } else { Some(x) }), + code: r.code, + }) + }) + .await + .map(|Val { lock, code }| (lock, code)) + } + + // ---------------------------------------------------------------------------------------------- + // impl `fs::Bundle` for `Val`. + + #[derive(Copy, Clone)] + pub enum Item { + Lock, + Code, + } + + impl fs::Item for Item { + fn path(&self, root: &Path) -> PathBuf { + match self { + Item::Lock => root.join("lock.txt"), + Item::Code => root.join("code.txt"), + } + } + } + + impl fs::Bundle for Val { + type Item = Item; + + fn items() -> &'static [Self::Item] { + &[Item::Lock, Item::Code] + } + + fn import(&mut self, item: Self::Item, data: Vec) -> error::Result<()> { + match item { + Item::Lock => self.lock = Some(String::from_utf8(data)?), + Item::Code => self.code = String::from_utf8(data)?, + } + Ok(()) + } + + fn export(&self, item: Self::Item) -> error::Result>> { + match item { + Item::Lock => Ok(self.lock.as_ref().map(|s| s.as_bytes().to_vec())), + Item::Code => Ok(Some(self.code.as_bytes().to_vec())), + } + } + } +} + mod fs { use super::*; diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 5b687b54fb..c8f8470b69 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -14,6 +14,7 @@ pub const ENTRYPOINT_OVERRIDE: &str = "_ENTRYPOINT_OVERRIDE"; pub const PREPROCESSOR_FAKE_ENTRYPOINT: &str = "__WM_PREPROCESSOR"; use crate::{ + apps::AppScriptId, error::{self, to_anyhow, Error}, flow_status::{FlowStatus, RestartedFrom}, flows::{FlowNodeId, FlowValue, Retry}, @@ -41,6 +42,7 @@ pub enum JobKind { DeploymentCallback, FlowScript, FlowNode, + AppScript, } #[derive(sqlx::FromRow, Debug, Serialize, Clone)] @@ -278,6 +280,12 @@ pub enum JobPayload { id: FlowNodeId, // flow_node(id). path: String, // flow node inner path (e.g. `outer/branchall-42`). }, + AppScript { + id: AppScriptId, // app_script(id). + path: Option, + language: ScriptLang, + cache_ttl: Option, + }, Code(RawCode), Dependencies { path: String, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 1e516f4821..c26a92872f 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3228,7 +3228,27 @@ pub async fn push<'c, 'd>( None, None, ) - } + }, + JobPayload::AppScript { + id, // app_script(id). + path, + language, + cache_ttl, + } => ( + Some(id.0), + path, + None, + JobKind::AppScript, + None, + None, + Some(language), + None, + None, + None, + cache_ttl, + None, + None, + ), JobPayload::ScriptHub { path } => { if path == "hub/7771/slack" || path == "hub/7836/slack" { permissioned_as = SUPERADMIN_NOTIFICATION_EMAIL.to_string(); @@ -3989,6 +4009,7 @@ pub async fn push<'c, 'd>( JobKind::DeploymentCallback => "jobs.run.deployment_callback", JobKind::FlowScript => "jobs.run.flow_script", JobKind::FlowNode => "jobs.run.flow_node", + JobKind::AppScript => "jobs.run.app_script", }; let audit_author = if format!("u/{user}") != permissioned_as && user != permissioned_as { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 47f2fb393f..238fd6a951 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -7,6 +7,7 @@ */ use windmill_common::{ + apps::AppScriptId, auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET}, scripts::PREVIEW_IS_TAR_CODEBASE_HASH, utils::WarnAfterExt, @@ -2311,6 +2312,20 @@ async fn handle_code_execution_job( codebase: None, } } + JobKind::AppScript => { + let (lockfile, content) = cache::app::fetch_script( + db, + AppScriptId(job.script_hash.unwrap_or(ScriptHash(0)).0), + ) + .await?; + ContentReqLangEnvs { + content, + lockfile, + language: job.language.to_owned(), + envs: None, + codebase: None, + } + } JobKind::DeploymentCallback => { get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await? } diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index ec739a88a2..ff1216da65 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -15,6 +15,7 @@ use windmill_common::jobs::JobPayload; use windmill_common::scripts::ScriptHash; use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file}; use windmill_common::{ + apps::AppScriptId, error::{self, to_anyhow}, flows::{add_virtual_items_if_necessary, FlowValue}, jobs::QueuedJob, @@ -648,7 +649,7 @@ pub async fn handle_flow_dependency_job( // Compute a lite version of the flow value (`RawScript` => `FlowScript`). let mut value_lite = flow.clone(); - tx = reduce( + tx = reduce_flow( tx, &mut value_lite.modules, &job_path, @@ -1053,6 +1054,41 @@ async fn insert_flow_node<'c>( Ok((tx, FlowNodeId(id))) } +async fn insert_app_script( + db: &sqlx::Pool, + app: i64, + code: String, + lock: Option, +) -> Result { + let code_sha256 = format!("{:x}", sha2::Sha256::digest(&code)); + let hash = { + let mut hasher = sha2::Sha256::new(); + hasher.update(app.to_le_bytes()); + hasher.update(&code_sha256); + hasher.update(lock.as_ref().unwrap_or(&Default::default())); + format!("{:x}", hasher.finalize()) + }; + + // Insert the app script if it doesn't exist. + sqlx::query_scalar!( + r#" + INSERT INTO app_script (app, hash, lock, code, code_sha256) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (hash) DO UPDATE SET app = EXCLUDED.app -- trivial update to return the id + RETURNING id + "#, + app, + hash, + lock, + code, + code_sha256 + ) + .fetch_one(db) + .await + .map(AppScriptId) + .map_err(Into::into) +} + async fn insert_flow_modules<'c>( mut tx: sqlx::Transaction<'c, sqlx::Postgres>, path: &str, @@ -1062,7 +1098,7 @@ async fn insert_flow_modules<'c>( modules: &mut Vec, modules_node: &mut Option, ) -> Result> { - tx = Box::pin(reduce( + tx = Box::pin(reduce_flow( tx, modules, path, @@ -1094,7 +1130,7 @@ async fn insert_flow_modules<'c>( Ok(tx) } -async fn reduce<'c>( +async fn reduce_flow<'c>( mut tx: sqlx::Transaction<'c, sqlx::Postgres>, modules: &mut Vec, path: &str, @@ -1107,7 +1143,7 @@ async fn reduce<'c>( let mut val = serde_json::from_str::(module.value.get()).map_err(|err| { Error::InternalErr(format!( - "reduce: Failed to parse flow module value: {}", + "reduce_flow: Failed to parse flow module value: {}", err )) })?; @@ -1203,6 +1239,41 @@ async fn reduce<'c>( Ok(tx) } +async fn reduce_app(db: &sqlx::Pool, value: &mut Value, app: i64) -> Result<()> { + match value { + Value::Object(object) => { + if let Some(Value::Object(script)) = object.get_mut("inlineScript") { + // replace `content` with an empty string: + let Some(Value::String(code)) = script.get_mut("content").map(std::mem::take) + else { + return Err(error::Error::InternalErr( + "Missing `content` in inlineScript".to_string(), + )); + }; + // remove `lock`: + let lock = script.remove("lock").and_then(|x| match x { + Value::String(s) => Some(s), + _ => None, + }); + let id = insert_app_script(db, app, code, lock).await?; + // insert the `id` into the `script` object: + script.insert("id".to_string(), json!(id.0)); + } else { + for (_, value) in object { + Box::pin(reduce_app(db, value, app)).await?; + } + } + } + Value::Array(array) => { + for value in array { + Box::pin(reduce_app(db, value, app)).await?; + } + } + _ => {} + } + Ok(()) +} + fn skip_creating_new_lock(language: &ScriptLang, content: &str) -> bool { if language == &ScriptLang::Bun || language == &ScriptLang::Bunnative { let anns = windmill_common::worker::TypeScriptAnnotations::parse(&content); @@ -1388,11 +1459,12 @@ pub async fn handle_app_dependency_job( .clone() .ok_or_else(|| Error::InternalErr("App Dependency requires script hash".to_owned()))? .0; - let value = sqlx::query_scalar!("SELECT value FROM app_version WHERE id = $1", id) + let record = sqlx::query!("SELECT app_id, value FROM app_version WHERE id = $1", id) .fetch_optional(db) - .await?; + .await? + .map(|record| (record.app_id, record.value)); - if let Some(value) = value { + if let Some((app_id, value)) = record { let value = lock_modules_app( value, job, @@ -1409,6 +1481,21 @@ pub async fn handle_app_dependency_job( ) .await?; + // Compute a lite version of the app value (w/ `inlineScript.{lock,code}`). + let mut value_lite = value.clone(); + reduce_app(db, &mut value_lite, app_id).await?; + if let Value::Object(object) = &mut value_lite { + object.insert("version".to_string(), json!(id)); + } + sqlx::query!( + "INSERT INTO app_version_lite (id, value) VALUES ($1, $2) + ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", + id, + sqlx::types::Json(to_raw_value(&value_lite)) as sqlx::types::Json>, + ) + .execute(db) + .await?; + // Re-check cancelation to ensure we don't accidentially override an app. if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id) .fetch_optional(db) diff --git a/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte b/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte index 9119282752..b135bc680a 100644 --- a/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte +++ b/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte @@ -377,11 +377,14 @@ : runnable if (inlineScript) { + if (inlineScript.id !== undefined) { + requestBody['id'] = inlineScript.id + } requestBody['raw_code'] = { - content: inlineScript.content, + content: inlineScript.id === undefined ? inlineScript.content : '', language: inlineScript.language ?? '', path: inlineScript.path, - lock: inlineScript.lock, + lock: inlineScript.id === undefined ? inlineScript.lock : undefined, cache_ttl: inlineScript.cache_ttl } } @@ -390,6 +393,10 @@ requestBody['path'] = runType !== 'hubscript' ? `${runType}/${path}` : `script/${path}` } + if ($app.version !== undefined) { + requestBody['version'] = $app.version + } + const uuid = await AppService.executeComponent({ workspace, path: defaultIfEmptyString($appPath, `u/${$userStore?.username ?? 'unknown'}/newapp`), diff --git a/frontend/src/lib/components/apps/types.ts b/frontend/src/lib/components/apps/types.ts index 5e6e8f00fa..7debc557f1 100644 --- a/frontend/src/lib/components/apps/types.ts +++ b/frontend/src/lib/components/apps/types.ts @@ -117,6 +117,7 @@ export type InlineScript = { cache_ttl?: number refreshOn?: { id: string; key: string }[] suggestedRefreshOn?: { id: string; key: string }[] + id?: number } export type AppCssItemName = 'viewer' | 'grid' | AppComponent['type'] @@ -163,6 +164,7 @@ export type App = { theme: AppTheme | undefined hideLegacyTopBar?: boolean | undefined mobileViewOnSmallerScreens?: boolean | undefined + version?: number } export type ConnectingInput = { diff --git a/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte index 1167ab1643..09bec77e67 100644 --- a/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte @@ -17,7 +17,7 @@ let can_write = false async function loadApp() { - app = await AppService.getAppByPath({ workspace: $workspaceStore!, path: $page.params.path }) + app = await AppService.getAppLiteByPath({ workspace: $workspaceStore!, path: $page.params.path }) can_write = canWrite(app?.path, app?.extra_perms!, $userStore) } From 691ef6468823c99ab129c53ae699de1a91f77c97 Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Fri, 6 Dec 2024 15:11:57 +0100 Subject: [PATCH 08/36] feat(frontend): render new job kinds (#4864) --- backend/windmill-api/openapi-deref.yaml | 6 ++++++ backend/windmill-api/openapi.yaml | 4 ++++ backend/windmill-worker/src/windmill-client.js | 6 ++++++ frontend/src/lib/components/runs/JobLoader.svelte | 7 +++++-- 4 files changed, 21 insertions(+), 2 deletions(-) diff --git a/backend/windmill-api/openapi-deref.yaml b/backend/windmill-api/openapi-deref.yaml index 0e7502d6cd..c8893f824c 100644 --- a/backend/windmill-api/openapi-deref.yaml +++ b/backend/windmill-api/openapi-deref.yaml @@ -8709,6 +8709,9 @@ paths: - identity - deploymentcallback - singlescriptflow + - flowscript + - flownode + - appscript schedule_path: type: string permissioned_as: @@ -9267,6 +9270,9 @@ paths: - identity - deploymentcallback - singlescriptflow + - flowscript + - flownode + - appscript schedule_path: type: string permissioned_as: diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index ccab668051..0bce856613 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -10983,6 +10983,8 @@ components: "deploymentcallback", "singlescriptflow", "flowscript", + "flownode", + "appscript", ] schedule_path: type: string @@ -11103,6 +11105,8 @@ components: "deploymentcallback", "singlescriptflow", "flowscript", + "flownode", + "appscript", ] schedule_path: type: string diff --git a/backend/windmill-worker/src/windmill-client.js b/backend/windmill-worker/src/windmill-client.js index 64281fede9..3ae15e7453 100644 --- a/backend/windmill-worker/src/windmill-client.js +++ b/backend/windmill-worker/src/windmill-client.js @@ -558,6 +558,9 @@ var $QueuedJob = { "identity", "deploymentcallback", "singlescriptflow", + "flowscript", + "flownode", + "appscript", ], }, schedule_path: { @@ -697,6 +700,9 @@ var $CompletedJob = { "identity", "deploymentcallback", "singlescriptflow", + "flowscript", + "flownode", + "appscript", ], }, schedule_path: { diff --git a/frontend/src/lib/components/runs/JobLoader.svelte b/frontend/src/lib/components/runs/JobLoader.svelte index d58a9d0af3..49d14f46a0 100644 --- a/frontend/src/lib/components/runs/JobLoader.svelte +++ b/frontend/src/lib/components/runs/JobLoader.svelte @@ -100,7 +100,10 @@ 'appdependencies', 'preview', 'flowpreview', - 'script_hub' + 'script_hub', + 'flowscript', + 'flownode', + 'appscript', ] return kinds.join(',') } else if (jobKindsCat == 'dependencies') { @@ -117,7 +120,7 @@ let kinds: CompletedJob['job_kind'][] = ['deploymentcallback'] return kinds.join(',') } else { - let kinds: CompletedJob['job_kind'][] = ['script', 'flow'] + let kinds: CompletedJob['job_kind'][] = ['script', 'flow', 'flowscript', 'flownode', 'appscript'] return kinds.join(',') } } From 8e9cb9c4d346281d116ce1bacc14017aacd5c366 Mon Sep 17 00:00:00 2001 From: HugoCasa Date: Fri, 6 Dec 2024 16:52:03 +0100 Subject: [PATCH 09/36] ee compute units telemetry (#4865) * ee compute units telemetry * update ee ref --- ...29dbf011f2390af9d02efc7112a32a84e1a1247a95973.json} | 10 ++++++++-- backend/ee-repo-ref.txt | 2 +- 2 files changed, 9 insertions(+), 3 deletions(-) rename backend/.sqlx/{query-a51007ca7b509b92faa6fe7aa59fa738594a2b7eaab8e38f3ca30e548f9a49a7.json => query-6a72df33cf12824c54b29dbf011f2390af9d02efc7112a32a84e1a1247a95973.json} (74%) diff --git a/backend/.sqlx/query-a51007ca7b509b92faa6fe7aa59fa738594a2b7eaab8e38f3ca30e548f9a49a7.json b/backend/.sqlx/query-6a72df33cf12824c54b29dbf011f2390af9d02efc7112a32a84e1a1247a95973.json similarity index 74% rename from backend/.sqlx/query-a51007ca7b509b92faa6fe7aa59fa738594a2b7eaab8e38f3ca30e548f9a49a7.json rename to backend/.sqlx/query-6a72df33cf12824c54b29dbf011f2390af9d02efc7112a32a84e1a1247a95973.json index 7e1f008dba..1521b9fd2f 100644 --- a/backend/.sqlx/query-a51007ca7b509b92faa6fe7aa59fa738594a2b7eaab8e38f3ca30e548f9a49a7.json +++ b/backend/.sqlx/query-6a72df33cf12824c54b29dbf011f2390af9d02efc7112a32a84e1a1247a95973.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT worker, worker_instance, vcpus, memory, ping_at, started_at FROM worker_ping WHERE ping_at > now() - interval '30 days' ORDER BY started_at", + "query": "SELECT worker, worker_instance, vcpus, memory, ping_at, started_at, worker_group FROM worker_ping WHERE ping_at > now() - interval '30 days' ORDER BY started_at", "describe": { "columns": [ { @@ -32,6 +32,11 @@ "ordinal": 5, "name": "started_at", "type_info": "Timestamptz" + }, + { + "ordinal": 6, + "name": "worker_group", + "type_info": "Varchar" } ], "parameters": { @@ -43,8 +48,9 @@ true, true, false, + false, false ] }, - "hash": "a51007ca7b509b92faa6fe7aa59fa738594a2b7eaab8e38f3ca30e548f9a49a7" + "hash": "6a72df33cf12824c54b29dbf011f2390af9d02efc7112a32a84e1a1247a95973" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index c7e750d0f4..6bb5c4686f 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -8606d98a692d11b09a387c5efbd6b4335c533fd3 \ No newline at end of file +5066da602260334767186e69ae5b6821feca0c71 \ No newline at end of file From 6d047449e200d51f060d4107ff5ad0e10af99a66 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 7 Dec 2024 16:29:20 +0100 Subject: [PATCH 10/36] feat: add otlp support (#4869) * all * all * add otel * add otel * update docker-image * update * update * update * update * update * update * update * update * update * update * update --- .github/workflows/docker-image.yml | 4 +- backend/Cargo.lock | 188 ++- backend/Cargo.toml | 9 +- backend/ee-repo-ref.txt | 2 +- backend/src/main.rs | 147 ++- backend/src/monitor.rs | 66 +- backend/tests/worker.rs | 5 +- backend/windmill-api/src/tracing_init.rs | 18 +- backend/windmill-common/Cargo.toml | 8 + .../windmill-common/src/global_settings.rs | 6 +- backend/windmill-common/src/lib.rs | 10 +- backend/windmill-common/src/otel_ee.rs | 54 + backend/windmill-common/src/tracing_init.rs | 94 +- backend/windmill-queue/Cargo.toml | 3 +- backend/windmill-queue/src/jobs.rs | 8 - backend/windmill-worker/Cargo.toml | 4 + backend/windmill-worker/src/common.rs | 10 +- backend/windmill-worker/src/handle_child.rs | 15 +- .../windmill-worker/src/result_processor.rs | 119 +- backend/windmill-worker/src/worker.rs | 153 ++- backend/windmill-worker/src/worker_flow.rs | 87 +- .../src/lib/components/AuthSettings.svelte | 255 ++++ .../src/lib/components/InstanceSetting.svelte | 783 ++++++++++++ .../lib/components/InstanceSettings.svelte | 1107 ++--------------- .../src/lib/components/OAuthSetting.svelte | 7 +- .../lib/components/OauthExtraParams.svelte | 2 +- .../src/lib/components/OauthScopes.svelte | 4 +- .../ObjectStoreConfigSettings.svelte | 269 ++-- .../lib/components/SuperadminSettings.svelte | 38 +- .../src/lib/components/instanceSettings.ts | 194 +-- 30 files changed, 2135 insertions(+), 1534 deletions(-) create mode 100644 backend/windmill-common/src/otel_ee.rs create mode 100644 frontend/src/lib/components/AuthSettings.svelte create mode 100644 frontend/src/lib/components/InstanceSetting.svelte diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index 42cc21264b..f1446210cc 100644 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -138,7 +138,7 @@ jobs: platforms: linux/amd64,linux/arm64 push: true build-args: | - features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka + features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka,otel tags: | ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }} ${{ steps.meta-ee-public.outputs.tags }} @@ -200,7 +200,7 @@ jobs: platforms: linux/amd64 push: true build-args: | - features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka + features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka,otel PYTHON_IMAGE=python:3.12.2-slim-bookworm tags: | ${{ steps.meta-ee-public-py312.outputs.tags }} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index dac85d4f09..e003c2300a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -4173,6 +4173,19 @@ dependencies = [ "tower-service", ] +[[package]] +name = "hyper-timeout" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3203a961e5c83b6f5498933e78b6b263e208c197b63e9c6c53cc82ffd3f63793" +dependencies = [ + "hyper 1.5.1", + "hyper-util", + "pin-project-lite", + "tokio", + "tower-service", +] + [[package]] name = "hyper-tls" version = "0.5.0" @@ -4809,7 +4822,7 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "674883a98273598ac3aad4301724c56734bea90574c5033af067e8f9fb5eb399" dependencies = [ - "prost", + "prost 0.12.6", "prost-types", ] @@ -5637,6 +5650,91 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "opentelemetry" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f3cebff57f7dbd1255b44d8bddc2cebeb0ea677dbaa2e25a3070a91b318f660" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "once_cell", + "pin-project-lite", + "thiserror 1.0.69", +] + +[[package]] +name = "opentelemetry-appender-tracing" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab5feffc321035ad94088a7e5333abb4d84a8726e54a802e736ce9dd7237e85b" +dependencies = [ + "opentelemetry", + "tracing", + "tracing-core", + "tracing-subscriber", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91cf61a1868dacc576bf2b2a1c3e9ab150af7272909e80085c3173384fe11f76" +dependencies = [ + "async-trait", + "futures-core", + "http 1.2.0", + "opentelemetry", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost 0.13.3", + "thiserror 1.0.69", + "tokio", + "tonic", + "tracing", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6e05acbfada5ec79023c85368af14abd0b307c015e9064d249b2a950ef459a6" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost 0.13.3", + "tonic", +] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc1b6902ff63b32ef6c489e8048c5e253e2e4a803ea3ea7e783914536eb15c52" + +[[package]] +name = "opentelemetry_sdk" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b742c1cae4693792cc564e58d75a2a0ba29421a34a85b50da92efa89ecb2bc" +dependencies = [ + "async-trait", + "futures-channel", + "futures-executor", + "futures-util", + "glob", + "once_cell", + "opentelemetry", + "percent-encoding", + "rand 0.8.5", + "serde_json", + "thiserror 1.0.69", + "tokio", + "tokio-stream", + "tracing", +] + [[package]] name = "option-ext" version = "0.2.0" @@ -6264,7 +6362,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "deb1435c188b76130da55f17a466d252ff7b1418b2ad3e037d127b94e3411f29" dependencies = [ "bytes", - "prost-derive", + "prost-derive 0.12.6", +] + +[[package]] +name = "prost" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b0487d90e047de87f984913713b85c601c05609aad5b0df4b4573fbf69aa13f" +dependencies = [ + "bytes", + "prost-derive 0.13.3", ] [[package]] @@ -6280,13 +6388,26 @@ dependencies = [ "syn 2.0.90", ] +[[package]] +name = "prost-derive" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9552f850d5f0964a4e4d0bf306459ac29323ddfbae05e35a7c0d35cb0803cc5" +dependencies = [ + "anyhow", + "itertools 0.13.0", + "proc-macro2", + "quote", + "syn 2.0.90", +] + [[package]] name = "prost-types" version = "0.12.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9091c90b0a32608e984ff2fa4091273cbdd755d54935c51d520887f4a1dbd5b0" dependencies = [ - "prost", + "prost 0.12.6", ] [[package]] @@ -9448,6 +9569,36 @@ dependencies = [ "winnow 0.6.20", ] +[[package]] +name = "tonic" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52" +dependencies = [ + "async-stream", + "async-trait", + "axum", + "base64 0.22.1", + "bytes", + "h2 0.4.7", + "http 1.2.0", + "http-body 1.0.1", + "http-body-util", + "hyper 1.5.1", + "hyper-timeout", + "hyper-util", + "percent-encoding", + "pin-project", + "prost 0.13.3", + "socket2", + "tokio", + "tokio-stream", + "tower 0.4.13", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "toolchain_find" version = "0.4.0" @@ -9469,11 +9620,16 @@ checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" dependencies = [ "futures-core", "futures-util", + "indexmap 1.9.3", "pin-project", "pin-project-lite", + "rand 0.8.5", + "slab", "tokio", + "tokio-util", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -9651,6 +9807,24 @@ dependencies = [ "url", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97a971f6058498b5c0f1affa23e7ea202057a7301dbff68e968b2d578bcbd053" +dependencies = [ + "js-sys", + "once_cell", + "opentelemetry", + "opentelemetry_sdk", + "smallvec", + "tracing", + "tracing-core", + "tracing-log 0.2.0", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-serde" version = "0.1.3" @@ -10644,6 +10818,11 @@ dependencies = [ "magic-crypt", "mail-send", "object_store", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-otlp", + "opentelemetry-semantic-conventions", + "opentelemetry_sdk", "pin-project-lite", "prometheus", "quick_cache", @@ -10662,6 +10841,7 @@ dependencies = [ "tracing-appender", "tracing-flame", "tracing-loki", + "tracing-opentelemetry", "tracing-subscriber", "uuid 1.11.0", "windmill-macros", @@ -10897,6 +11077,7 @@ dependencies = [ "hmac", "itertools 0.13.0", "lazy_static", + "opentelemetry", "prometheus", "regex", "reqwest 0.12.9", @@ -10961,6 +11142,7 @@ dependencies = [ "object_store", "once_cell", "openidconnect", + "opentelemetry", "pem 3.0.4", "postgres-native-tls", "prometheus", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 3f591dfd84..4646af9f5e 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -63,6 +63,7 @@ tantivy = ["dep:windmill-indexer", "windmill-api/tantivy"] sqlx = ["windmill-worker/sqlx"] deno_core = ["windmill-worker/deno_core", "dep:deno_core"] kafka = ["windmill-api/kafka"] +otel = ["windmill-common/otel", "windmill-worker/otel"] [dependencies] anyhow.workspace = true @@ -95,7 +96,6 @@ deno_core = { workspace = true, optional = true } object_store = { workspace = true, optional = true } quote.workspace = true - [target.'cfg(not(target_env = "msvc"))'.dependencies] tikv-jemallocator = { optional = true, workspace = true } tikv-jemalloc-sys = { optional = true, workspace = true } @@ -279,6 +279,13 @@ tar = "^0" http = "^1" async-stream = "^0" +opentelemetry = "0.27.0" +tracing-opentelemetry = "0.28.0" +opentelemetry_sdk = { version = "*", features = ["rt-tokio"] } +opentelemetry-otlp = "0.27.0" +opentelemetry-appender-tracing = "0.27.0" +opentelemetry-semantic-conventions = "*" + tikv-jemallocator = { version = "0.5" } tikv-jemalloc-sys = { version = "^0.5" } tikv-jemalloc-ctl = { version = "^0.5" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 6bb5c4686f..b5181c39f2 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -5066da602260334767186e69ae5b6821feca0c71 \ No newline at end of file +ad89b5a1566159490eceb59c3e5c9416d9aa855c \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index d403ad212a..80edff1ac9 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -8,7 +8,7 @@ use anyhow::Context; use monitor::{ - reload_delete_logs_periodically_setting, reload_indexer_config, + load_otel, reload_delete_logs_periodically_setting, reload_indexer_config, reload_timeout_wait_result_setting, send_current_log_file_to_object_store, send_logs_to_object_store, }; @@ -37,7 +37,7 @@ use windmill_common::{ EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING, - OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, + OAUTH_SETTING, OTEL_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TIMEOUT_WAIT_RESULT_SETTING, }, @@ -221,11 +221,66 @@ async fn windmill_main() -> anyhow::Result<()> { let hostname = hostname(); - #[cfg(not(feature = "flamegraph"))] - let _guard = windmill_common::tracing_init::initialize_tracing(&hostname); + let mut enable_standalone_indexer: bool = false; + + let mode = std::env::var("MODE") + .map(|x| x.to_lowercase()) + .map(|x| { + if &x == "server" { + println!("Binary is in 'server' mode"); + Mode::Server + } else if &x == "worker" { + tracing::info!("Binary is in 'worker' mode"); + #[cfg(windows)] + { + println!("It is highly recommended to use the agent mode instead on windows (MODE=agent) and to pass a BASE_INTERNAL_URL"); + } + Mode::Worker + } else if &x == "agent" { + println!("Binary is in 'agent' mode"); + if std::env::var("BASE_INTERNAL_URL").is_err() { + panic!("BASE_INTERNAL_URL is required in agent mode") + } + if std::env::var("JOB_TOKEN").is_err() { + println!("JOB_TOKEN is not passed, hence workers will still need to create permissions for each job and the DATABASE_URL needs to be of a role that can INSERT into the job_perms table") + } + + #[cfg(not(feature = "enterprise"))] + { + panic!("Agent mode is only available in the EE, ignoring..."); + } + #[cfg(feature = "enterprise")] + Mode::Agent + } else if &x == "indexer" { + tracing::info!("Binary is in 'indexer' mode"); + #[cfg(not(feature = "tantivy"))] + { + eprintln!("Cannot start the indexer because tantivy is not included in this binary/image. Make sure you are using the EE image if you want to access the full text search features."); + panic!("Indexer mode requires compiling with the tantivy feature flag."); + } + #[cfg(feature = "tantivy")] + Mode::Indexer + } else if &x == "standalone+search"{ + enable_standalone_indexer = true; + println!("Binary is in 'standalone' mode with search enabled"); + Mode::Standalone + } + else { + if &x != "standalone" { + eprintln!("mode not recognized, defaulting to standalone: {x}"); + } else { + println!("Binary is in 'standalone' mode"); + } + Mode::Standalone + } + }) + .unwrap_or_else(|_| { + tracing::info!("Mode not specified, defaulting to standalone"); + Mode::Standalone + }); #[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))] - tracing::info!("jemalloc enabled"); + println!("jemalloc enabled"); #[cfg(feature = "flamegraph")] let _guard = windmill_common::tracing_init::setup_flamegraph(); @@ -236,13 +291,13 @@ async fn windmill_main() -> anyhow::Result<()> { "cache" => { #[cfg(feature = "embedding")] { - tracing::info!("Caching embedding model..."); + println!("Caching embedding model..."); windmill_api::embeddings::ModelInstance::load_model_files().await?; - tracing::info!("Cached embedding model"); + println!("Cached embedding model"); } #[cfg(not(feature = "embedding"))] { - tracing::warn!("Embeddings are not enabled, ignoring..."); + println!("Embeddings are not enabled, ignoring..."); } cache_hub_scripts(std::env::args().nth(2)).await?; @@ -256,64 +311,6 @@ async fn windmill_main() -> anyhow::Result<()> { _ => {} } - let mut enable_standalone_indexer: bool = false; - - let mode = std::env::var("MODE") - .map(|x| x.to_lowercase()) - .map(|x| { - if &x == "server" { - tracing::info!("Binary is in 'server' mode"); - Mode::Server - } else if &x == "worker" { - tracing::info!("Binary is in 'worker' mode"); - #[cfg(windows)] - { - tracing::warn!("It is highly recommended to use the agent mode instead on windows (MODE=agent) and to pass a BASE_INTERNAL_URL"); - } - Mode::Worker - } else if &x == "agent" { - tracing::info!("Binary is in 'agent' mode"); - if std::env::var("BASE_INTERNAL_URL").is_err() { - panic!("BASE_INTERNAL_URL is required in agent mode") - } - if std::env::var("JOB_TOKEN").is_err() { - tracing::warn!("JOB_TOKEN is not passed, hence workers will still need to create permissions for each job and the DATABASE_URL needs to be of a role that can INSERT into the job_perms table") - } - - #[cfg(not(feature = "enterprise"))] - { - panic!("Agent mode is only available in the EE, ignoring..."); - } - #[cfg(feature = "enterprise")] - Mode::Agent - } else if &x == "indexer" { - tracing::info!("Binary is in 'indexer' mode"); - #[cfg(not(feature = "tantivy"))] - { - tracing::error!("Cannot start the indexer because tantivy is not included in this binary/image. Make sure you are using the EE image if you want to access the full text search features."); - panic!("Indexer mode requires compiling with the tantivy feature flag."); - } - #[cfg(feature = "tantivy")] - Mode::Indexer - } else if &x == "standalone+search"{ - enable_standalone_indexer = true; - tracing::info!("Binary is in 'standalone' mode with search enabled"); - Mode::Standalone - } - else { - if &x != "standalone" { - tracing::error!("mode not recognized, defaulting to standalone: {x}"); - } else { - tracing::info!("Binary is in 'standalone' mode"); - } - Mode::Standalone - } - }) - .unwrap_or_else(|_| { - tracing::info!("Mode not specified, defaulting to standalone"); - Mode::Standalone - }); - #[allow(unused_mut)] let mut num_workers = if mode == Mode::Server || mode == Mode::Indexer { 0 @@ -325,7 +322,7 @@ async fn windmill_main() -> anyhow::Result<()> { }; if num_workers > 1 { - tracing::warn!( + println!( "We STRONGLY recommend using at most 1 worker per container, use at your own risks" ); } @@ -347,10 +344,15 @@ async fn windmill_main() -> anyhow::Result<()> { IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)) }; - tracing::info!("Connecting to database..."); + println!("Connecting to database..."); let db = windmill_common::connect_db(server_mode, indexer_mode).await?; + + load_otel(&db).await; tracing::info!("Database connected"); + #[cfg(not(feature = "flamegraph"))] + let _guard = windmill_common::tracing_init::initialize_tracing(&hostname, &mode); + let num_version = sqlx::query_scalar!("SELECT version()").fetch_one(&db).await; tracing::info!( @@ -776,6 +778,15 @@ Windmill Community Edition {GIT_VERSION} tracing::error!(error = %e, "Could not reload debug metrics setting"); } }, + OTEL_SETTING => { + tracing::info!("OTEL setting changed, restarting"); + // we wait a bit randomly to avoid having all servers and workers shutdown at same time + let rd_delay = rand::thread_rng().gen_range(0..4); + tokio::time::sleep(Duration::from_secs(rd_delay)).await; + if let Err(e) = tx.send(()) { + tracing::error!(error = %e, "Could not send killpill"); + } + }, REQUEST_SIZE_LIMIT_SETTING => { if server_mode { tracing::info!("Request limit size change detected, killing server expecting to be restarted"); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index df3b695168..262927cce3 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -12,7 +12,7 @@ use std::{ use chrono::{NaiveDateTime, Utc}; use futures::{stream::FuturesUnordered, StreamExt}; -use serde::de::DeserializeOwned; +use serde::{de::DeserializeOwned, Deserializer}; use sqlx::{Pool, Postgres}; use tokio::{ join, @@ -40,7 +40,7 @@ use windmill_common::{ EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, - PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, + OTEL_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING, }, @@ -58,7 +58,8 @@ use windmill_common::{ }, BASE_URL, CRITICAL_ALERT_MUTE_UI_ENABLED, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS, METRICS_DEBUG_ENABLED, METRICS_ENABLED, - MONITOR_LOGS_ON_OBJECT_STORE, SERVICE_LOG_RETENTION_SECS, + MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, + SERVICE_LOG_RETENTION_SECS, }; use windmill_queue::cancel_job; use windmill_worker::{ @@ -199,6 +200,65 @@ pub async fn load_metrics_enabled(db: &DB) -> error::Result<()> { Ok(()) } +fn empty_string_as_none<'de, D>(deserializer: D) -> Result, D::Error> +where + D: Deserializer<'de>, +{ + let option = as serde::Deserialize>::deserialize(deserializer)?; + Ok(option.filter(|s| !s.is_empty())) +} + +#[derive(serde::Deserialize)] +struct OtelSetting { + metrics_enabled: Option, + logs_enabled: Option, + tracing_enabled: Option, + #[serde(default, deserialize_with = "empty_string_as_none")] + otel_exporter_otlp_endpoint: Option, + #[serde(default, deserialize_with = "empty_string_as_none")] + otel_exporter_otlp_headers: Option, + #[serde(default, deserialize_with = "empty_string_as_none")] + otel_exporter_otlp_protocol: Option, + #[serde(default, deserialize_with = "empty_string_as_none")] + otel_exporter_otlp_compression: Option, +} + +pub async fn load_otel(db: &DB) { + let otel = load_value_from_global_settings(db, OTEL_SETTING).await; + if let Ok(v) = otel { + if let Some(v) = v { + let deser = serde_json::from_value::(v); + if let Ok(o) = deser { + let metrics_enabled = o.metrics_enabled.unwrap_or(false); + let logs_enabled = o.logs_enabled.unwrap_or(false); + let tracing_enabled = o.tracing_enabled.unwrap_or(false); + + OTEL_METRICS_ENABLED.store(metrics_enabled, Ordering::Relaxed); + OTEL_LOGS_ENABLED.store(logs_enabled, Ordering::Relaxed); + OTEL_TRACING_ENABLED.store(tracing_enabled, Ordering::Relaxed); + if let Some(endpoint) = o.otel_exporter_otlp_endpoint.as_ref() { + std::env::set_var("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint); + } + if let Some(headers) = o.otel_exporter_otlp_headers.as_ref() { + std::env::set_var("OTEL_EXPORTER_OTLP_HEADERS", headers); + } + if let Some(protocol) = o.otel_exporter_otlp_protocol { + std::env::set_var("OTEL_EXPORTER_OTLP_PROTOCOL", protocol); + } + if let Some(compression) = o.otel_exporter_otlp_compression { + std::env::set_var("OTEL_EXPORTER_OTLP_COMPRESSION", compression); + } + tracing::info!("OTEL settings loaded: tracing ({tracing_enabled}), logs ({logs_enabled}), metrics ({metrics_enabled}), endpoint ({:?}), headers defined: ({})", + o.otel_exporter_otlp_endpoint, o.otel_exporter_otlp_headers.is_some()); + } else { + tracing::error!("Error deserializing otel settings"); + } + } + } else { + tracing::error!("Error loading otel settings: {}", otel.unwrap_err()); + } +} + pub async fn load_tag_per_workspace_enabled(db: &DB) -> error::Result<()> { let metrics_enabled = load_value_from_global_settings(db, DEFAULT_TAGS_PER_WORKSPACE_SETTING).await; diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index a229450651..2aebf6f5e8 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -77,7 +77,10 @@ async fn initialize_tracing() { static ONCE: Once = Once::new(); ONCE.call_once(|| { - let _ = windmill_common::tracing_init::initialize_tracing("test"); + let _ = windmill_common::tracing_init::initialize_tracing( + "test", + &windmill_common::utils::Mode::Standalone, + ); }); } diff --git a/backend/windmill-api/src/tracing_init.rs b/backend/windmill-api/src/tracing_init.rs index 60dab73810..c12cbe3192 100644 --- a/backend/windmill-api/src/tracing_init.rs +++ b/backend/windmill-api/src/tracing_init.rs @@ -29,11 +29,19 @@ impl OnResponse for MyOnResponse { _span: &tracing::Span, ) { if *LOG_REQUESTS { - tracing::info!( - latency = latency.as_millis(), - status = response.status().as_u16(), - "response" - ) + if response.status().is_success() { + tracing::info!( + latency = latency.as_millis(), + status = response.status().as_u16(), + "response" + ) + } else { + tracing::error!( + latency = latency.as_millis(), + status = response.status().as_u16(), + "response" + ) + } } } } diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index b56b9a39cf..5b9a74c361 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -13,6 +13,7 @@ flamegraph = ["dep:tracing-flame"] loki = ["dep:tracing-loki"] benchmark = [] parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"] +otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk", "dep:opentelemetry", "dep:tracing-opentelemetry", "dep:opentelemetry-appender-tracing"] [lib] name = "windmill_common" @@ -64,5 +65,12 @@ croner = "2.0.6" quick_cache.workspace = true pin-project-lite.workspace = true +opentelemetry-semantic-conventions = { workspace = true, optional = true } +opentelemetry-otlp = { workspace = true, optional = true } +opentelemetry_sdk = { workspace = true, optional = true } +opentelemetry = { workspace = true, optional = true } +tracing-opentelemetry = { workspace = true, optional = true } +opentelemetry-appender-tracing = { workspace = true, optional = true } + [target.'cfg(not(target_env = "msvc"))'.dependencies] tikv-jemalloc-ctl = { optional = true, workspace = true } diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 7b932d8f50..9dbfd03b0e 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -35,8 +35,9 @@ pub const CRITICAL_ALERT_MUTE_UI_SETTING: &str = "critical_alert_mute_ui"; pub const DEV_INSTANCE_SETTING: &str = "dev_instance"; pub const JWT_SECRET_SETTING: &str = "jwt_secret"; pub const EMAIL_DOMAIN_SETTING: &str = "email_domain"; +pub const OTEL_SETTING: &str = "otel"; -pub const ENV_SETTINGS: [&str; 51] = [ +pub const ENV_SETTINGS: [&str; 54] = [ "DISABLE_NSJAIL", "MODE", "NUM_WORKERS", @@ -88,4 +89,7 @@ pub const ENV_SETTINGS: [&str; 51] = [ "WORKER_GROUP", "SAML_METADATA", "INSTANCE_IS_DEV", + "OTEL_METRICS", + "OTEL_TRACING", + "OTEL_LOGS", ]; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index ad47b198af..4140f6e58b 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -36,20 +36,20 @@ pub mod job_s3_helpers_ee; pub mod jobs; pub mod more_serde; pub mod oauth2; +pub mod otel_ee; pub mod queue; pub mod s3_helpers; pub mod schedule; pub mod scripts; pub mod server; pub mod stats_ee; +pub mod tracing_init; pub mod users; pub mod utils; pub mod variables; pub mod worker; pub mod workspaces; -pub mod tracing_init; - pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50; pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5; pub const DEFAULT_MAX_CONNECTIONS_INDEXER: u32 = 5; @@ -87,6 +87,12 @@ lazy_static::lazy_static! { .unwrap_or_else(|| SocketAddr::from(([0, 0, 0, 0], *METRICS_PORT))); pub static ref METRICS_ENABLED: AtomicBool = AtomicBool::new(std::env::var("METRICS_PORT").is_ok() || std::env::var("METRICS_ADDR").is_ok()); + + pub static ref OTEL_METRICS_ENABLED: AtomicBool = AtomicBool::new(std::env::var("OTEL_METRICS").is_ok()); + pub static ref OTEL_TRACING_ENABLED: AtomicBool = AtomicBool::new(std::env::var("OTEL_TRACING").is_ok()); + pub static ref OTEL_LOGS_ENABLED: AtomicBool = AtomicBool::new(std::env::var("OTEL_LOGS").is_ok()); + + pub static ref METRICS_DEBUG_ENABLED: AtomicBool = AtomicBool::new(false); pub static ref CRITICAL_ALERT_MUTE_UI_ENABLED: AtomicBool = AtomicBool::new(false); diff --git a/backend/windmill-common/src/otel_ee.rs b/backend/windmill-common/src/otel_ee.rs new file mode 100644 index 0000000000..f310885c0b --- /dev/null +++ b/backend/windmill-common/src/otel_ee.rs @@ -0,0 +1,54 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +use crate::{jobs::QueuedJob, utils::Mode}; +use uuid::Uuid; + +pub fn set_span_parent(_span: &tracing::Span, _rj: &Uuid) {} + +#[cfg(not(all(feature = "otel", feature = "enterprise")))] +pub(crate) type OtelProvider = Option<()>; + +#[cfg(all(feature = "otel", feature = "enterprise"))] +pub(crate) type OtelProvider = Option; + +#[cfg(not(feature = "otel"))] +pub fn otel_ctx() -> () {} + +#[cfg(feature = "otel")] +#[inline(always)] +pub fn otel_ctx() -> opentelemetry::Context { + opentelemetry::Context::current() +} + +#[cfg(not(feature = "otel"))] +impl FutureExt for T {} + +#[cfg(not(feature = "otel"))] +pub trait FutureExt: Sized { + fn with_context(self, _otel_cx: ()) -> Self { + self + } +} + +use tracing_subscriber::EnvFilter; + +pub(crate) fn init_logs_bridge(_mode: &Mode) -> Option { + None +} + +#[cfg(all(feature = "otel", feature = "enterprise"))] +pub(crate) fn init_otlp_tracer(_mode: &Mode) -> Option { + None +} + +pub(crate) fn init_meter_provider(_mode: &Mode) -> OtelProvider { + None +} + +pub fn add_root_flow_job_to_otlp(_queued_job: &QueuedJob, _success: bool) {} diff --git a/backend/windmill-common/src/tracing_init.rs b/backend/windmill-common/src/tracing_init.rs index 13b345e36e..af34540add 100644 --- a/backend/windmill-common/src/tracing_init.rs +++ b/backend/windmill-common/src/tracing_init.rs @@ -7,13 +7,23 @@ */ use const_format::concatcp; + +use std::{ + collections::HashMap, + sync::{Arc, RwLock}, +}; +use tracing::Event; use tracing_appender::non_blocking::{NonBlockingBuilder, WorkerGuard}; +use tracing_subscriber::layer::Context; use tracing_subscriber::{ + filter::Targets, fmt::{format, Layer}, prelude::*, EnvFilter, }; +use crate::utils::Mode; + fn json_layer() -> Layer> { tracing_subscriber::fmt::layer() .json() @@ -34,7 +44,10 @@ pub const LOGS_SERVICE: &str = "logs/services/"; pub const TMP_WINDMILL_LOGS_SERVICE: &str = concatcp!("/tmp/windmill/", LOGS_SERVICE); -pub fn initialize_tracing(hostname: &str) -> WorkerGuard { +pub fn initialize_tracing( + hostname: &str, + mode: &Mode, +) -> (WorkerGuard, crate::otel_ee::OtelProvider) { let style = std::env::var("RUST_LOG_STYLE").unwrap_or_else(|_| "auto".into()); if std::env::var("RUST_LOG").is_ok_and(|x| x == "debug" || x == "info") { @@ -44,7 +57,17 @@ pub fn initialize_tracing(hostname: &str) -> WorkerGuard { ) } - let env_filter = EnvFilter::from_default_env(); + let meter_provider = crate::otel_ee::init_meter_provider(mode); + + #[cfg(all(feature = "otel", feature = "enterprise"))] + let opentelemetry = crate::otel_ee::init_otlp_tracer(mode) + .map(|x| tracing_opentelemetry::layer().with_tracer(x)); + + #[cfg(not(all(feature = "otel", feature = "enterprise")))] + let opentelemetry: Option = None; + + let logs_bridge = crate::otel_ee::init_logs_bridge(&mode); + use tracing_appender::rolling::{RollingFileAppender, Rotation}; let log_dir = format!("{}/{}/", TMP_WINDMILL_LOGS_SERVICE, hostname); @@ -61,39 +84,59 @@ pub fn initialize_tracing(hostname: &str) -> WorkerGuard { .finish(file_appender); let stdout_and_log_file_writer = std::io::stdout.and(log_file_writer); - let ts_base = tracing_subscriber::registry().with(env_filter); + // let job_logs_filter = tracing_subscriber::filter::Targets::new() + // .with_target("windmill:job_log", tracing::Level::TRACE); - #[cfg(feature = "loki")] - let ts_base = { - let (layer, task) = tracing_loki::builder() - .build_url(reqwest::Url::parse("http://127.0.0.1:3100").unwrap()) - .expect("build loki url"); - tokio::spawn(task); - ts_base.with(layer) - }; + let env_filter = EnvFilter::builder() + .with_default_directive(tracing::level_filters::LevelFilter::ERROR.into()) + .from_env_lossy(); + + let ts_base = tracing_subscriber::registry().with(env_filter); match *JSON_FMT { true => ts_base + .with(logs_bridge) + .with(opentelemetry) + // .with(env_filter2.add_directive("windmill:job_log=off".parse().unwrap())) .with( json_layer() .with_writer(stdout_and_log_file_writer) - .flatten_event(true), + .flatten_event(true) + .with_filter( + Targets::new() + .with_target( + "windmill:job_log", + tracing::level_filters::LevelFilter::OFF, + ) + .with_default(tracing::level_filters::LevelFilter::INFO), + ), ) .with(CountingLayer::new()) .init(), false => ts_base + .with(logs_bridge) + .with(opentelemetry) + // .with(env_filter2.add_directive("windmill:job_log=off".parse().unwrap())) .with( compact_layer() .with_writer(stdout_and_log_file_writer) .with_ansi(style.to_lowercase() != "never") .with_file(true) .with_line_number(true) - .with_target(false), + .with_target(false) + .with_filter( + Targets::new() + .with_target( + "windmill:job_log", + tracing::level_filters::LevelFilter::OFF, + ) + .with_default(tracing::level_filters::LevelFilter::INFO), + ), ) .with(CountingLayer::new()) .init(), } - _guard + (_guard, meter_provider) } #[cfg(feature = "flamegraph")] @@ -112,13 +155,6 @@ pub fn setup_flamegraph() -> impl Drop { _guard } -use std::{ - collections::HashMap, - sync::{Arc, RwLock}, -}; -use tracing::Event; -use tracing_subscriber::layer::Context; - lazy_static::lazy_static! { pub static ref LOG_COUNTING_BY_MIN: Arc>> = Arc::new(RwLock::new(HashMap::new())); } @@ -144,22 +180,6 @@ impl CountingLayer { } } -// impl CountingLayer { -// pub fn new() -> Self { -// CountingLayer { counter: Arc::new(Mutex::new(LogCounter::new())) } -// } - -// pub fn get_counts(&self) -> (usize, usize) { -// let counter = self.counter.lock().unwrap(); -// (counter.non_error_count, counter.error_count) -// } - -// pub fn reset_counts(&self) { -// let mut counter = self.counter.lock().unwrap(); -// counter.reset(); -// } -// } - pub const LOG_TIMESTAMP_FMT: &str = "%Y-%m-%d-%H-%M"; impl tracing_subscriber::Layer for CountingLayer diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index 7ac744a885..8f94de918a 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -43,4 +43,5 @@ bigdecimal.workspace = true axum.workspace = true serde_urlencoded.workspace = true regex.workspace = true -backon.workspace = true \ No newline at end of file +backon.workspace = true +opentelemetry.workspace = true diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index c26a92872f..ef37834400 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -32,7 +32,6 @@ use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction}; #[cfg(feature = "benchmark")] use std::time::Instant; use tokio::{sync::RwLock, time::sleep}; -use tracing::{instrument, Instrument}; use ulid::Ulid; use uuid::Uuid; use windmill_audit::audit_ee::{audit_log, AuditAuthor}; @@ -454,7 +453,6 @@ where } } -#[instrument(level = "trace", skip_all)] pub async fn add_completed_job_error( db: &Pool, queued_job: &QueuedJob, @@ -510,7 +508,6 @@ lazy_static::lazy_static! { pub static ref GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE: Option = std::env::var("GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE").ok(); } -#[instrument(level = "trace", skip_all, name = "add_completed_job")] pub async fn add_completed_job( db: &Pool, queued_job: &QueuedJob, @@ -643,9 +640,7 @@ pub async fn add_completed_job( .fetch_one(&mut *tx) .await .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?; - // tracing::error!("2 {:?}", start.elapsed()); - // add_time!(bench, "add_completed_job query END"); if !queued_job.is_flow_step { if _duration > 500 @@ -1259,7 +1254,6 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> Ok(()) } -#[instrument(level = "trace", skip_all)] pub async fn handle_maybe_scheduled_job<'c>( db: &Pool, job: &QueuedJob, @@ -2429,7 +2423,6 @@ async fn extract_result_from_job_result( } } -#[instrument(level = "trace", skip_all)] pub async fn delete_job<'c>( mut tx: Transaction<'c, Postgres>, w_id: &str, @@ -4039,7 +4032,6 @@ pub async fn push<'c, 'd>( script_path.as_ref().map(|x| x.as_str()), Some(hm), ) - .instrument(tracing::info_span!("job_run", email = &email)) .await?; } diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index cb64757a7f..22640e577a 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -19,6 +19,7 @@ flow_testing = [] cloud = [] sqlx = [] deno_core = ["dep:deno_fetch", "dep:deno_webidl", "dep:deno_web", "dep:deno_net", "dep:deno_console", "dep:deno_url", "dep:deno_core", "dep:deno_ast", "dep:deno_tls"] +otel = ["windmill-common/otel", "dep:opentelemetry"] [dependencies] windmill-queue.workspace = true @@ -92,6 +93,9 @@ yaml-rust.workspace = true swc_ecma_parser.workspace = true backon.workspace = true +opentelemetry = { workspace = true, optional = true } + + [build-dependencies] deno_fetch = { workspace = true, optional = true } deno_webidl = { workspace = true, optional = true } diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 31d8ecc4dd..f36954819d 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -866,11 +866,11 @@ fn tentatively_improve_error(err: Error, executable: &str) -> Error { #[cfg(windows)] let err_msg = "program not found"; - if err - .to_string() - .contains(&err_msg) - { - return Error::InternalErr(format!("Executable {executable} not found on worker. PATH: {}", *PATH_ENV)); + if err.to_string().contains(&err_msg) { + return Error::InternalErr(format!( + "Executable {executable} not found on worker. PATH: {}", + *PATH_ENV + )); } return err; } diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 4d09c36abc..cb35070a40 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -87,7 +87,7 @@ async fn kill_process_tree(pid: Option) -> Result<(), String> { /// - update the `last_line` and `logs` strings with the program output /// - update "queue"."last_ping" every five seconds /// - kill process if we exceed timeout or "queue"."canceled" is set -#[tracing::instrument(level = "trace", skip_all)] +#[tracing::instrument(name="run_subprocess", level = "info", skip_all, fields(otel.name = %child_name))] pub async fn handle_child( job_id: &Uuid, db: &Pool, @@ -435,8 +435,6 @@ pub async fn handle_child( } } - - async fn get_mem_peak(pid: Option, nsjail: bool) -> i32 { if pid.is_none() { return -1; @@ -692,7 +690,10 @@ fn child_joined_output_stream( let stdout = BufReader::new(stdout).lines(); let stderr = BufReader::new(stderr).lines(); - stream::select(lines_to_stream(stderr, true), lines_to_stream(stdout, false)) + stream::select( + lines_to_stream(stderr, true), + lines_to_stream(stdout, false), + ) } pub fn lines_to_stream( @@ -702,14 +703,10 @@ pub fn lines_to_stream( stream::poll_fn(move |cx| { std::pin::Pin::new(&mut lines) .poll_next_line(cx) - .map(|result| { - process_streaming_log_lines(result, stderr) - }) + .map(|result| process_streaming_log_lines(result, stderr)) }) } - - pub fn process_status(status: ExitStatus) -> error::Result<()> { if status.success() { Ok(()) diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 25bfade2e8..a897459224 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -1,3 +1,6 @@ +#[cfg(feature = "otel")] +use opentelemetry::trace::FutureExt; + use serde::Serialize; use sqlx::{types::Json, Pool, Postgres}; use std::{ @@ -7,6 +10,9 @@ use std::{ Arc, }, }; +use tracing::{field, Instrument}; +#[cfg(not(feature = "otel"))] +use windmill_common::otel_ee::FutureExt; use uuid::Uuid; @@ -85,7 +91,49 @@ pub fn start_background_processor( JobKind::Dependencies | JobKind::FlowDependencies ); - handle_receive_completed_job( + let success = jc.success; + + let span = tracing::span!( + tracing::Level::INFO, + "job_postprocessing", + job_id = %jc.job.id, root_job = field::Empty, workspace_id = %jc.job.workspace_id, worker = %worker_name,tag = %jc.job.tag, + // hostname = %hostname, + language = field::Empty, + script_path = field::Empty, + flow_step_id = field::Empty, + parent_job = field::Empty, + otel.name = field::Empty + ); + let rj = if let Some(root_job) = jc.job.root_job { + root_job + } else { + jc.job.id + }; + windmill_common::otel_ee::set_span_parent(&span, &rj); + + if let Some(lg) = jc.job.language.as_ref() { + span.record("language", lg.as_str()); + } + if let Some(step_id) = jc.job.flow_step_id.as_ref() { + span.record( + "otel.name", + format!("job_postprocessing {}", step_id).as_str(), + ); + span.record("flow_step_id", step_id.as_str()); + } else { + span.record("otel.name", "job postprocessing"); + } + if let Some(parent_job) = jc.job.parent_job.as_ref() { + span.record("parent_job", parent_job.to_string().as_str()); + } + if let Some(script_path) = jc.job.script_path.as_ref() { + span.record("script_path", script_path.as_str()); + } + if let Some(root_job) = jc.job.root_job.as_ref() { + span.record("root_job", root_job.to_string().as_str()); + } + + let root_job = handle_receive_completed_job( jc, &base_internal_url, &db, @@ -96,8 +144,14 @@ pub fn start_background_processor( #[cfg(feature = "benchmark")] &mut bench, ) + .instrument(span) .await; + if let Some(root_job) = root_job { + windmill_common::otel_ee::add_root_flow_job_to_otlp(&root_job, success); + tracing::error!(job_id = %root_job.id, parent_job = ?root_job.parent_job, "ADDDED root job completed"); + } + if is_init_script_and_failure { tracing::error!("init script errored, exiting"); killpill_tx.send(()).unwrap_or_default(); @@ -199,7 +253,11 @@ async fn send_job_completed( token, duration, }; - job_completed_tx.send(jc).await.expect("send job completed") + job_completed_tx + .send(jc) + .with_context(windmill_common::otel_ee::otel_ctx()) + .await + .expect("send job completed") } pub async fn process_result( @@ -271,6 +329,7 @@ pub async fn process_result( token, duration, ) + .with_context(windmill_common::otel_ee::otel_ctx()) .await; Ok(true) } @@ -315,6 +374,7 @@ pub async fn process_result( token, duration, ) + .with_context(windmill_common::otel_ee::otel_ctx()) .await; Ok(false) } @@ -330,7 +390,7 @@ pub async fn handle_receive_completed_job( worker_name: &str, job_completed_tx: Sender, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, -) { +) -> Option> { let token = jc.token.clone(); let workspace = jc.job.workspace_id.clone(); let client = AuthedClient { @@ -342,7 +402,7 @@ pub async fn handle_receive_completed_job( let job = jc.job.clone(); let mem_peak = jc.mem_peak.clone(); let canceled_by = jc.canceled_by.clone(); - if let Err(err) = process_completed_job( + match process_completed_job( jc, &client, db, @@ -355,26 +415,29 @@ pub async fn handle_receive_completed_job( ) .await { - handle_job_error( - db, - &client, - job.as_ref(), - mem_peak, - canceled_by, - err, - false, - same_worker_tx.clone(), - &worker_dir, - worker_name, - job_completed_tx, - #[cfg(feature = "benchmark")] - bench, - ) - .await; + Err(err) => { + handle_job_error( + db, + &client, + job.as_ref(), + mem_peak, + canceled_by, + err, + false, + same_worker_tx.clone(), + &worker_dir, + worker_name, + job_completed_tx, + #[cfg(feature = "benchmark")] + bench, + ) + .await; + None + } + Ok(r) => r, } } -#[tracing::instrument(name = "completed_job", level = "info", skip_all, fields(job_id = %job.id))] pub async fn process_completed_job( JobCompleted { job, result, mem_peak, success, cached_res_path, canceled_by, duration, .. }: JobCompleted, client: &AuthedClient, @@ -384,7 +447,7 @@ pub async fn process_completed_job( worker_name: &str, job_completed_tx: Sender, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, -) -> windmill_common::error::Result<()> { +) -> windmill_common::error::Result>> { if success { // println!("bef completed job{:?}", SystemTime::now()); if let Some(cached_path) = cached_res_path { @@ -414,8 +477,8 @@ pub async fn process_completed_job( if is_flow_step { if let Some(parent_job) = parent_job { - tracing::info!(parent_flow = %parent_job, subflow = %job_id, "updating flow status (2)"); - update_flow_status_after_job_completion( + // tracing::info!(parent_flow = %parent_job, subflow = %job_id, "updating flow status (2)"); + let r = update_flow_status_after_job_completion( db, client, parent_job, @@ -434,9 +497,10 @@ pub async fn process_completed_job( ) .warn_after_seconds(10) .await?; + add_time!(bench, "updated flow status END"); + return Ok(r); } } - add_time!(bench, "updated flow status END"); } else { let result = add_completed_job_error( db, @@ -454,7 +518,7 @@ pub async fn process_completed_job( if job.is_flow_step { if let Some(parent_job) = job.parent_job { tracing::error!(parent_flow = %parent_job, subflow = %job.id, "process completed job error, updating flow status"); - update_flow_status_after_job_completion( + let r = update_flow_status_after_job_completion( db, client, parent_job, @@ -473,10 +537,11 @@ pub async fn process_completed_job( ) .warn_after_seconds(10) .await?; + return Ok(r); } } } - Ok(()) + return Ok(None); } #[tracing::instrument(name = "job_error", level = "info", skip_all, fields(job_id = %job.id))] diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 238fd6a951..f179396123 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -6,6 +6,10 @@ * LICENSE-AGPL for a copy of the license. */ +// #[cfg(feature = "otel")] +// use opentelemetry::{global, KeyValue}; + + use windmill_common::{ apps::AppScriptId, auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET}, @@ -25,7 +29,7 @@ use const_format::concatcp; #[cfg(feature = "prometheus")] use prometheus::IntCounter; -use tracing::Instrument; +use tracing::{field, Instrument}; #[cfg(feature = "prometheus")] use windmill_common::METRICS_DEBUG_ENABLED; #[cfg(feature = "prometheus")] @@ -88,30 +92,12 @@ use tokio::{ use rand::Rng; use crate::{ - ansible_executor::handle_ansible_job, - bash_executor::{handle_bash_job, handle_powershell_job}, - bun_executor::handle_bun_job, - common::{ + ansible_executor::handle_ansible_job, bash_executor::{handle_bash_job, handle_powershell_job}, bun_executor::handle_bun_job, common::{ build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args, update_worker_ping_for_failed_init_script, OccupancyMetrics, - }, - deno_executor::handle_deno_job, - go_executor::handle_go_job, - graphql_executor::do_graphql, - handle_child::SLOW_LOGS, - handle_job_error, - job_logger::NO_LOGS_AT_ALL, - js_eval::{eval_fetch_timeout, transpile_ts}, - mysql_executor::do_mysql, - pg_executor::do_postgresql, - php_executor::handle_php_job, - python_executor::handle_python_job, - result_processor::{process_result, start_background_processor}, - rust_executor::handle_rust_job, - worker_flow::{handle_flow, update_flow_status_in_progress, Step}, - worker_lockfiles::{ + }, deno_executor::handle_deno_job, go_executor::handle_go_job, graphql_executor::do_graphql, handle_child::SLOW_LOGS, handle_job_error, job_logger::NO_LOGS_AT_ALL, js_eval::{eval_fetch_timeout, transpile_ts}, mysql_executor::do_mysql, pg_executor::do_postgresql, php_executor::handle_php_job, python_executor::handle_python_job, result_processor::{process_result, start_background_processor}, rust_executor::handle_rust_job, worker_flow::{handle_flow, update_flow_status_in_progress, Step}, worker_lockfiles::{ handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job, - }, + } }; use backon::ConstantBuilder; @@ -720,7 +706,10 @@ fn add_outstanding_wait_time( }.in_current_span()); } -#[tracing::instrument(name = "worker", level = "info", skip_all, fields(worker = %worker_name, hostname = %hostname))] +// struct WorkerMtrics { +// job_ +// } + pub async fn run_worker( db: &Pool, hostname: &str, @@ -736,6 +725,7 @@ pub async fn run_worker( #[cfg(not(feature = "enterprise"))] if !*DISABLE_NSJAIL { tracing::warn!( + worker = %worker_name, hostname = %hostname, "NSJAIL to sandbox process in untrusted environments is an enterprise feature but allowed to be used for testing purposes" ); } @@ -743,10 +733,10 @@ pub async fn run_worker( let start_time = Instant::now(); let worker_dir = format!("{TMP_DIR}/{worker_name}"); - tracing::debug!(worker_dir = %worker_dir, "Creating worker dir"); + tracing::debug!(worker = %worker_name, hostname = %hostname, worker_dir = %worker_dir, "Creating worker dir"); if let Some(ref netrc) = *NETRC { - tracing::info!("Writing netrc at {}/.netrc", HOME_ENV.as_str()); + tracing::info!(worker = %worker_name, hostname = %hostname, "Writing netrc at {}/.netrc", HOME_ENV.as_str()); write_file(&HOME_ENV, ".netrc", netrc).expect("could not write netrc"); } @@ -961,6 +951,16 @@ pub async fn run_worker( None }; + + // let worker_resource = &[ + // KeyValue::new("hostname", hostname.to_string()), + // KeyValue::new("worker", worker_name.to_string()), + // ]; + // // Create a meter from the above MeterProvider. + // let meter = global::meter("windmill"); + // let counter = meter.u64_counter("jobs.execution").build(); + + let mut occupancy_metrics = OccupancyMetrics::new(start_time); let mut jobs_executed = 0; @@ -1016,6 +1016,7 @@ pub async fn run_worker( IS_READY.store(true, Ordering::Relaxed); tracing::info!( + worker = %worker_name, hostname = %hostname, "listening for jobs, WORKER_GROUP: {}, config: {:?}", *WORKER_GROUP, WORKER_CONFIG.read().await @@ -1051,7 +1052,7 @@ pub async fn run_worker( if i_worker == 1 { if let Err(e) = queue_init_bash_maybe(db, same_worker_tx.clone(), &worker_name).await { killpill_tx.send(()).unwrap_or_default(); - tracing::error!("Error queuing init bash script for worker {worker_name}: {e:#}"); + tracing::error!(worker = %worker_name, hostname = %hostname, "Error queuing init bash script for worker {worker_name}: {e:#}"); return; } } @@ -1084,7 +1085,7 @@ pub async fn run_worker( #[cfg(feature = "enterprise")] { if let Ok(_) = killpill_rx.try_recv() { - tracing::info!("killpill received on worker waiting for valid key"); + tracing::info!(worker = %worker_name, hostname = %hostname, "killpill received on worker waiting for valid key"); job_completed_tx .0 .send(SendResult::Kill) @@ -1096,6 +1097,7 @@ pub async fn run_worker( if !valid_key { tracing::error!( + worker = %worker_name, hostname = %hostname, "Invalid license key, workers require a valid license key, sleeping for 30s waiting for valid key to be set" ); tokio::time::sleep(Duration::from_secs(10)).await; @@ -1109,7 +1111,7 @@ pub async fn run_worker( #[cfg(feature = "prometheus")] if let Some(wk) = worker_busy.as_ref() { wk.set(0); - tracing::debug!("set worker busy to 0"); + tracing::debug!(worker = %worker_name, hostname = %hostname, "set worker busy to 0"); } occupancy_metrics.running_job_started_at = None; @@ -1121,7 +1123,7 @@ pub async fn run_worker( .try_into() .unwrap(), ); - tracing::debug!("set uptime metric"); + tracing::debug!(worker = %worker_name, hostname = %hostname, "set uptime metric"); } if last_ping.elapsed().as_secs() > NUM_SECS_PING { @@ -1165,15 +1167,19 @@ pub async fn run_worker( ) .notify(|err, dur| { tracing::error!( + worker = %worker_name, hostname = %hostname, "retrying updating worker ping in {dur:#?}, err: {err:#?}" ); }) .sleep(tokio::time::sleep) .await { - tracing::error!("failed to update worker ping, exiting: {}", e); + tracing::error!( + worker = %worker_name, hostname = %hostname, + "failed to update worker ping, exiting: {}", e); killpill_tx.send(()).unwrap_or_default(); } tracing::info!( + worker = %worker_name, hostname = %hostname, "ping update, memory: container={}MB, windmill={}MB", memory_usage.unwrap_or_default() / (1024 * 1024), wm_memory_usage.unwrap_or_default() / (1024 * 1024) @@ -1185,16 +1191,18 @@ pub async fn run_worker( if (jobs_executed as u32 + vacuum_shift) % VACUUM_PERIOD == 0 { let db2 = db.clone(); let current_span = tracing::Span::current(); + let worker_name = worker_name.clone(); + let hostname = hostname.to_string(); tokio::task::spawn( (async move { - tracing::info!("vacuuming queue and completed_job"); + tracing::info!(worker = %worker_name, hostname = %hostname, "vacuuming queue"); if let Err(e) = sqlx::query!("VACUUM (skip_locked) queue") .execute(&db2) .await { - tracing::error!("failed to vacuum queue: {}", e); + tracing::error!(worker = %worker_name, hostname = %hostname, "failed to vacuum queue: {}", e); } - tracing::info!("vacuumed queue and completed_job"); + tracing::info!(worker = %worker_name, hostname = %hostname, "vacuumed queue"); }) .instrument(current_span), ); @@ -1230,6 +1238,7 @@ pub async fn run_worker( if let Ok(same_worker_job) = same_worker_rx.try_recv() { same_worker_queue_size.fetch_sub(1, Ordering::SeqCst); tracing::debug!( + worker = %worker_name, hostname = %hostname, "received {} from same worker channel", same_worker_job.job_id ); @@ -1242,6 +1251,7 @@ pub async fn run_worker( .map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string())); if r.is_err() && !same_worker_job.recoverable { tracing::error!( + worker = %worker_name, hostname = %hostname, "failed to fetch same_worker job on a non recoverable job, exiting" ); job_completed_tx @@ -1255,7 +1265,7 @@ pub async fn run_worker( } } else if let Ok(_) = killpill_rx.try_recv() { if !killed_but_draining_same_worker_jobs { - tracing::info!("received killpill for worker {}, jobs are not pulled anymore except same_worker jobs", i_worker); + tracing::info!(worker = %worker_name, hostname = %hostname, "received killpill for worker {}, jobs are not pulled anymore except same_worker jobs", i_worker); killed_but_draining_same_worker_jobs = true; job_completed_tx .0 @@ -1266,10 +1276,10 @@ pub async fn run_worker( continue; } else if killed_but_draining_same_worker_jobs { if job_completed_processor_is_done.load(Ordering::SeqCst) { - tracing::info!("all running jobs have completed and all completed jobs have been fully processed, exiting"); + tracing::info!(worker = %worker_name, hostname = %hostname, "all running jobs have completed and all completed jobs have been fully processed, exiting"); break; } else { - tracing::info!("there may be same_worker jobs to process later, waiting for job_completed_processor to finish progressing all remaining flows before exiting"); + tracing::info!(worker = %worker_name, hostname = %hostname, "there may be same_worker jobs to process later, waiting for job_completed_processor to finish progressing all remaining flows before exiting"); tokio::time::sleep(Duration::from_millis(200)).await; continue; } @@ -1294,7 +1304,7 @@ pub async fn run_worker( if !agent_mode && duration_pull_s > 0.5 { let empty = job.as_ref().is_ok_and(|x| x.0.is_none()); - tracing::warn!("pull took more than 0.5s ({duration_pull_s}), this is a sign that the database is VERY undersized for this load. empty: {empty}, err: {err_pull}"); + tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.5s ({duration_pull_s}), this is a sign that the database is VERY undersized for this load. empty: {empty}, err: {err_pull}"); #[cfg(feature = "prometheus")] if empty { if let Some(wp) = worker_pull_over_500_counter_empty.as_ref() { @@ -1305,7 +1315,7 @@ pub async fn run_worker( } } else if !agent_mode && duration_pull_s > 0.1 { let empty = job.as_ref().is_ok_and(|x| x.0.is_none()); - tracing::warn!("pull took more than 0.1s ({duration_pull_s}) this is a sign that the database is undersized for this load. empty: {empty}, err: {err_pull}"); + tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.1s ({duration_pull_s}) this is a sign that the database is undersized for this load. empty: {empty}, err: {err_pull}"); #[cfg(feature = "prometheus")] if empty { if let Some(wp) = worker_pull_over_100_counter_empty.as_ref() { @@ -1359,7 +1369,7 @@ pub async fn run_worker( last_executed_job = None; jobs_executed += 1; - tracing::debug!("started handling of job {}", job.id); + tracing::debug!(worker = %worker_name, hostname = %hostname, "started handling of job {}", job.id); if matches!(job.job_kind, JobKind::Script | JobKind::Preview) { if !dedicated_workers.is_empty() { @@ -1424,6 +1434,11 @@ pub async fn run_worker( ) .await; + // counter.add( + // 1, + // worker_resource + // ); + #[cfg(feature = "prometheus")] let _timer = register_metric( &WORKER_EXECUTION_DURATION, @@ -1511,6 +1526,40 @@ pub async fn run_worker( let PulledJob { job, raw_code, raw_lock, raw_flow } = job; let arc_job = Arc::new(job); add_time!(bench, "handle_queued_job START"); + + + let span = tracing::span!(tracing::Level::INFO, "job", + job_id = %arc_job.id, root_job = field::Empty, workspace_id = %arc_job.workspace_id, worker = %worker_name, hostname = %hostname, tag = %arc_job.tag, + language = field::Empty, + script_path = field::Empty, flow_step_id = field::Empty, parent_job = field::Empty, + otel.name = field::Empty); + let rj = if let Some(root_job) = arc_job.root_job { + root_job + } else { + arc_job.id + }; + if let Some(lg) = arc_job.language.as_ref() { + span.record("language", lg.as_str()); + } + if let Some(step_id) = arc_job.flow_step_id.as_ref() { + span.record("otel.name", format!("job {}", step_id).as_str()); + span.record("flow_step_id", step_id.as_str()); + } else { + span.record("otel.name", "job"); + } + if let Some(parent_job) = arc_job.parent_job.as_ref() { + span.record("parent_job", parent_job.to_string().as_str()); + } + if let Some(script_path) = arc_job.script_path.as_ref() { + span.record("script_path", script_path.as_str()); + } + if let Some(root_job) = arc_job.root_job.as_ref() { + span.record("root_job", root_job.to_string().as_str()); + } + + windmill_common::otel_ee::set_span_parent(&span, &rj); + // span.context().span().add_event_with_timestamp("job created".to_string(), arc_job.created_at.into(), vec![]); + match handle_queued_job( arc_job.clone(), raw_code, @@ -1529,6 +1578,7 @@ pub async fn run_worker( #[cfg(feature = "benchmark")] &mut bench, ) + .instrument(span) .await { Err(err) => { @@ -1568,6 +1618,8 @@ pub async fn run_worker( _ => {} } + + #[cfg(feature = "prometheus")] if let Some(duration) = _timer.map(|x| x.stop_and_record()) { register_metric( @@ -1607,7 +1659,7 @@ pub async fn run_worker( if let Some(secs) = *EXIT_AFTER_NO_JOB_FOR_SECS { if let Some(lj) = last_executed_job { if lj.elapsed().as_secs() > secs { - tracing::info!("no job for {} seconds, exiting", secs); + tracing::info!(worker = %worker_name, hostname = %hostname, "no job for {} seconds, exiting", secs); break; } } else { @@ -1638,12 +1690,12 @@ pub async fn run_worker( }); } Err(err) => { - tracing::error!("Failed to pull jobs: {}", err); + tracing::error!(worker = %worker_name, hostname = %hostname, "Failed to pull jobs: {}", err); } }; } - tracing::info!("worker {} exiting", worker_name); + tracing::info!(worker = %worker_name, hostname = %hostname, "worker {} exiting", worker_name); #[cfg(feature = "benchmark")] { @@ -1658,22 +1710,24 @@ pub async fn run_worker( if has_dedicated_workers { for handle in dedicated_handles { if let Err(e) = handle.await { - tracing::error!("error in dedicated worker waiting for it to end: {:?}", e) + tracing::error!(worker = %worker_name, hostname = %hostname, "error in dedicated worker waiting for it to end: {:?}", e) } } - tracing::info!("all dedicated workers have exited"); + tracing::info!(worker = %worker_name, hostname = %hostname, "all dedicated workers have exited"); } drop(job_completed_tx); - tracing::info!("waiting for job_completed_processor to finish processing remaining jobs"); + tracing::info!(worker = %worker_name, hostname = %hostname, "waiting for job_completed_processor to finish processing remaining jobs"); if let Err(e) = send_result.await { tracing::error!("error in awaiting send_result process: {e:?}") } - tracing::info!("worker {} exited", worker_name); - tracing::info!("number of jobs executed: {}", jobs_executed); + tracing::info!(worker = %worker_name, hostname = %hostname, "worker {} exited", worker_name); + tracing::info!(worker = %worker_name, hostname = %hostname, "number of jobs executed: {}", jobs_executed); } + + async fn queue_init_bash_maybe<'c>( db: &Pool, same_worker_tx: SameWorkerSender, @@ -1798,7 +1852,6 @@ pub struct PreviousResult<'a> { pub previous_result: Option<&'a RawValue>, } -#[tracing::instrument(name = "job", level = "info", skip_all, fields(job_id = %job.id))] async fn handle_queued_job( job: Arc, raw_code: Option, @@ -1816,6 +1869,9 @@ async fn handle_queued_job( occupancy_metrics: &mut OccupancyMetrics, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) -> windmill_common::error::Result { + // Extract the active span from the context + + if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job))); } @@ -2159,6 +2215,7 @@ async fn handle_queued_job( } } + pub fn build_envs( envs: Option>, ) -> windmill_common::error::Result> { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 74949d0bcf..b5770166c5 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -77,45 +77,34 @@ pub async fn update_flow_status_after_job_completion( worker_name: &str, job_completed_tx: Sender, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, -) -> error::Result<()> { +) -> error::Result>> { // this is manual tailrecursion because async_recursion blows up the stack - // todo!(); potentially_crash_for_testing(); - let mut rec = update_flow_status_after_job_completion_internal( - db, - client, + let mut rec = RecUpdateFlowStatusAfterJobCompletion { flow, - job_id_for_status, - w_id, + job_id_for_status: job_id_for_status.clone(), success, result, - unrecoverable, - same_worker_tx.clone(), - worker_dir, stop_early_override, - false, - worker_name, - job_completed_tx.clone(), - #[cfg(feature = "benchmark")] - bench, - ) - .await?; - while let Some(nrec) = rec { + skip_error_handler: false, + }; + let mut unrecoverable = unrecoverable; + loop { potentially_crash_for_testing(); - rec = match update_flow_status_after_job_completion_internal( + let nrec = match update_flow_status_after_job_completion_internal( db, client, - nrec.flow, - &nrec.job_id_for_status, + rec.flow, + &rec.job_id_for_status, w_id, - nrec.success, - nrec.result, - false, + rec.success, + rec.result, + unrecoverable, same_worker_tx.clone(), worker_dir, - nrec.stop_early_override, - nrec.skip_error_handler, + rec.stop_early_override, + rec.skip_error_handler, worker_name, job_completed_tx.clone(), #[cfg(feature = "benchmark")] @@ -125,12 +114,12 @@ pub async fn update_flow_status_after_job_completion( { Ok(j) => j, Err(e) => { - tracing::error!("Error while updating flow status of {} after completion of {}, updating flow status again with error: {e:#}", nrec.flow,&nrec.job_id_for_status); + tracing::error!("Error while updating flow status of {} after completion of {}, updating flow status again with error: {e:#}", rec.flow, &rec.job_id_for_status); update_flow_status_after_job_completion_internal( db, client, - nrec.flow, - &nrec.job_id_for_status, + rec.flow, + &rec.job_id_for_status, w_id, false, Arc::new(to_raw_value(&Json(&WrappedError { @@ -139,8 +128,8 @@ pub async fn update_flow_status_after_job_completion( true, same_worker_tx.clone(), worker_dir, - nrec.stop_early_override, - nrec.skip_error_handler, + rec.stop_early_override, + rec.skip_error_handler, worker_name, job_completed_tx.clone(), #[cfg(feature = "benchmark")] @@ -148,9 +137,33 @@ pub async fn update_flow_status_after_job_completion( ) .await? } + }; + unrecoverable = false; + match nrec { + UpdateFlowStatusAfterJobCompletion::Done(job) => { + add_time!(bench, "update flow status internal END"); + return Ok(Some(job)); + } + UpdateFlowStatusAfterJobCompletion::Rec(nrec) => { + rec = nrec; + }, + UpdateFlowStatusAfterJobCompletion::NonLastParallelBranch => { + add_time!(bench, "update flow status internal END"); + return Ok(None); + }, + UpdateFlowStatusAfterJobCompletion::NotDone => { + add_time!(bench, "update flow status internal END"); + return Ok(None); + } } } - Ok(()) +} + +pub enum UpdateFlowStatusAfterJobCompletion { + Rec(RecUpdateFlowStatusAfterJobCompletion), + Done(Arc), + NotDone, + NonLastParallelBranch, } pub struct RecUpdateFlowStatusAfterJobCompletion { flow: uuid::Uuid, @@ -188,7 +201,7 @@ pub async fn update_flow_status_after_job_completion_internal( worker_name: &str, job_completed_tx: Sender, #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, -) -> error::Result> { +) -> error::Result { add_time!(bench, "update flow status internal START"); let ( should_continue_flow, @@ -603,7 +616,7 @@ pub async fn update_flow_status_after_job_completion_internal( ); } add_time!(bench, "non final parallel flow finished"); - return Ok(None); + return Ok(UpdateFlowStatusAfterJobCompletion::NonLastParallelBranch); } } FlowStatusModule::InProgress { @@ -1148,7 +1161,7 @@ pub async fn update_flow_status_after_job_completion_internal( if let Some(parent_job) = flow_job.parent_job { tracing::info!(subflow_id = %flow_job.id, parent_id = %parent_job, "subflow is finished, updating parent flow status"); - return Ok(Some(RecUpdateFlowStatusAfterJobCompletion { + return Ok(UpdateFlowStatusAfterJobCompletion::Rec(RecUpdateFlowStatusAfterJobCompletion { flow: parent_job, job_id_for_status: flow, success: success && !is_failure_step, @@ -1162,9 +1175,9 @@ pub async fn update_flow_status_after_job_completion_internal( })); } } - Ok(None) + Ok(UpdateFlowStatusAfterJobCompletion::Done(flow_job)) } else { - Ok(None) + Ok(UpdateFlowStatusAfterJobCompletion::NotDone) } } diff --git a/frontend/src/lib/components/AuthSettings.svelte b/frontend/src/lib/components/AuthSettings.svelte new file mode 100644 index 0000000000..56f6d625de --- /dev/null +++ b/frontend/src/lib/components/AuthSettings.svelte @@ -0,0 +1,255 @@ + + +
+ + SSO + OAuth + SCIM/SAML + +
+ +
+ {#if tab === 'sso'} + {#if !$enterpriseLicense || $enterpriseLicense.endsWith('_pro')} + + Without EE, the number of SSO users is limited to 10. SCIM/SAML is available on EE + + {/if} + +
+
+ When at least one of the below options is set, users will be able to login to Windmill via + their third-party account. +
To test SSO, the recommended workflow is to to save the settings and try to login in + an incognito window. + Learn more
+
+
+ + + + + + + + + + + + {#each Object.keys(oauths) as k} + {#if !['authelia', 'authentik', 'google', 'microsoft', 'github', 'gitlab', 'jumpcloud', 'okta', 'keycloak', 'slack', 'kanidm', 'zitadel'].includes(k) && 'login_config' in oauths[k]} + {#if oauths[k]} +
+
+ + + { + delete oauths[k] + oauths = { ...oauths } + }} + /> +
+
+ + + + {#if !windmillBuiltins.includes(k) && k != 'slack'} + + {/if} +
+
+ {/if} + {/if} + {/each} +
+
+ + +
+
+ +
+ {:else if tab === 'oauth'} +
+ When one of the below options is set, you will be able to create a specific resource + containing a token automatically generated by the third-party provider. +
+ To test it after setting an oauth client, go to the Resources menu and create a new one of the + type of your oauth client (i.e. a 'github' resource if you set Github OAuth). +
Learn more
+
+
+ +
+ + {#each Object.keys(oauths) as k} + {#if oauths[k] && !('login_config' in oauths[k])} + {#if !['slack'].includes(k) && oauths[k]} +
+
+ + + { + delete oauths[k] + oauths = { ...oauths } + }} + /> +
+
+ + + {#if !windmillBuiltins.includes(k) && k != 'slack'} + + {/if} + {#if k == 'snowflake_oauth'} + + {/if} +
+
+ {/if} + {/if} + {/each} + +
+ + {#if oauth_name == 'custom'} + + {:else} + + {/if} + +
+ {:else if tab == 'scim'} + + {/if} +
diff --git a/frontend/src/lib/components/InstanceSetting.svelte b/frontend/src/lib/components/InstanceSetting.svelte new file mode 100644 index 0000000000..1bac352d5b --- /dev/null +++ b/frontend/src/lib/components/InstanceSetting.svelte @@ -0,0 +1,783 @@ + + +{#if (!setting.cloudonly || isCloudHosted()) && showSetting(setting.key, $values) && !(setting.hiddenIfNull && $values[setting.key] == null)} + {#if setting.ee_only != undefined && !$enterpriseLicense} +
+ + EE only {#if setting.ee_only != ''}{setting.ee_only}{/if} +
+ {/if} + +