Merge remote-tracking branch 'upstream/main' into appeditor-multiselect-input-type

This commit is contained in:
Clement Zhang
2024-11-14 10:35:35 +01:00
130 changed files with 5557 additions and 1620 deletions
@@ -0,0 +1,58 @@
name: Build windows executable for this branch
on:
workflow_dispatch:
env:
CARGO_INCREMENTAL: 0
SQLX_OFFLINE: true
DISABLE_EMBEDDING: true
RUST_LOG: info
jobs:
cargo_build_windows:
runs-on: windows-latest
steps:
- uses: actions/checkout@v4
- name: Read EE repo commit hash
shell: pwsh
run: |
$ee_repo_ref = Get-Content .\backend\ee-repo-ref.txt
echo "ee_repo_ref=$ee_repo_ref" | Out-File -FilePath $env:GITHUB_ENV -Append
- name: Checkout windmill-ee-private repository
uses: actions/checkout@v4
with:
repository: windmill-labs/windmill-ee-private
path: ./windmill-ee-private
ref: ${{ env.ee_repo_ref }}
token: ${{ secrets.WINDMILL_EE_PRIVATE_ACCESS }}
fetch-depth: 0
- name: Substitute EE code
shell: bash
run: |
./backend/substitute_ee_code.sh --copy --dir ./windmill-ee-private
- name: Cargo build windows
timeout-minutes: 90
run: |
vcpkg.exe install openssl-windows:x64-windows
vcpkg.exe install openssl:x64-windows-static
vcpkg.exe integrate install
$env:VCPKGRS_DYNAMIC=1
$env:OPENSSL_DIR="${Env:VCPKG_INSTALLATION_ROOT}\installed\x64-windows-static"
mkdir frontend/build && cd backend
New-Item -Path . -Name "windmill-api/openapi-deref.yaml" -ItemType "File" -Force
cargo build --release --features=enterprise,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core
- name: Rename binary with corresponding architecture
run: |
Rename-Item -Path ".\backend\target\release\windmill.exe" -NewName "windmill-ee.exe"
- name: Upload artifact
uses: actions/upload-artifact@v3
with:
name: windmill-ee-binary
path: ./backend/target/release/windmill-ee.exe
+83
View File
@@ -1,5 +1,88 @@
# Changelog
## [1.423.2](https://github.com/windmill-labs/windmill/compare/v1.423.1...v1.423.2) (2024-11-13)
### Bug Fixes
* fix intempestive expr type change in flow transform ([84bd06e](https://github.com/windmill-labs/windmill/commit/84bd06e90650e949c11e2b09447a7bad1ab60a95))
## [1.423.1](https://github.com/windmill-labs/windmill/compare/v1.423.0...v1.423.1) (2024-11-12)
### Bug Fixes
* **autoscaling:** autoscaling thresholds to be >= and not > ([de00944](https://github.com/windmill-labs/windmill/commit/de00944f09699ff33a382f5a14f515ccf90b2454))
## [1.423.0](https://github.com/windmill-labs/windmill/compare/v1.422.1...v1.423.0) (2024-11-12)
### Features
* s3 input available for public apps ([#4685](https://github.com/windmill-labs/windmill/issues/4685)) ([1671005](https://github.com/windmill-labs/windmill/commit/167100510032ab53cd609fb2c7629e67faceb093))
### Bug Fixes
* prevent underflow for autoscaling scalein ([5e9870a](https://github.com/windmill-labs/windmill/commit/5e9870a8a993032ec7d45c62e0023008da42c74a))
* support multiple pip-extra-index-url with commas ([50861fc](https://github.com/windmill-labs/windmill/commit/50861fccc0cd86d4c76120c3306adf94f375330e))
## [1.422.1](https://github.com/windmill-labs/windmill/compare/v1.422.0...v1.422.1) (2024-11-12)
### Bug Fixes
* fix password inputs ([e5e174a](https://github.com/windmill-labs/windmill/commit/e5e174ae9516f4c6b94ceb6e258b467f5c9a1f1a))
## [1.422.0](https://github.com/windmill-labs/windmill/compare/v1.421.2...v1.422.0) (2024-11-11)
### Features
* expandable subflows in flows ([#4683](https://github.com/windmill-labs/windmill/issues/4683)) ([d44976f](https://github.com/windmill-labs/windmill/commit/d44976f35e45ade510d1ec220b5a1503e11f3db9))
* **frontend:** critical alerts UI ([#4653](https://github.com/windmill-labs/windmill/issues/4653)) ([d9148ea](https://github.com/windmill-labs/windmill/commit/d9148eaa78680a93d81d71847a7df67e01f3c110))
## [1.421.2](https://github.com/windmill-labs/windmill/compare/v1.421.1...v1.421.2) (2024-11-08)
### Bug Fixes
* **bash:** correctly propagate exit errors ([8bc9a02](https://github.com/windmill-labs/windmill/commit/8bc9a021a88391147c9170f56b5a0edddd55bc7d))
## [1.421.1](https://github.com/windmill-labs/windmill/compare/v1.421.0...v1.421.1) (2024-11-08)
### Bug Fixes
* **python-client:** fix small break params of write_s3_file ([3ea12f1](https://github.com/windmill-labs/windmill/commit/3ea12f1821500e8b84549b892e3e1bb56a6ace4b))
## [1.421.0](https://github.com/windmill-labs/windmill/compare/v1.420.1...v1.421.0) (2024-11-07)
### Features
* http custom routes for static assets ([#4666](https://github.com/windmill-labs/windmill/issues/4666)) ([dc8bd6d](https://github.com/windmill-labs/windmill/commit/dc8bd6d2b5d6f5cd8521f9034853175ec78d5639))
### Bug Fixes
* improve nested schema editor field change ([a3feca7](https://github.com/windmill-labs/windmill/commit/a3feca719799ea2bf08f2b49350e6b732a24abf4))
## [1.420.1](https://github.com/windmill-labs/windmill/compare/v1.420.0...v1.420.1) (2024-11-07)
### Bug Fixes
* improve prop filtering on flow prop picker ([0501ad8](https://github.com/windmill-labs/windmill/commit/0501ad8a99f6f7d9fd26c996b754e8afe8f958b1))
## [1.420.0](https://github.com/windmill-labs/windmill/compare/v1.419.0...v1.420.0) (2024-11-07)
### Features
* **frontend:** detect expr in flow input transform + filter right panel based on expr ([#4651](https://github.com/windmill-labs/windmill/issues/4651)) ([e9b7dca](https://github.com/windmill-labs/windmill/commit/e9b7dca20387e775fa50aaecd832890251582cf9))
* **frontend:** nodes from flow can be connected directly in expr input through a plug icon ([#4652](https://github.com/windmill-labs/windmill/issues/4652)) ([ceaf56c](https://github.com/windmill-labs/windmill/commit/ceaf56c21ee2e548bc93859f9e0303e53b25b241))
## [1.419.0](https://github.com/windmill-labs/windmill/compare/v1.418.0...v1.419.0) (2024-11-06)
+2
View File
@@ -355,6 +355,8 @@ you to have it being synced automatically everyday.
See the [./frontend/README_DEV.md](./frontend/README_DEV.md) file for all
running options.
Using [Nix](./frontend/README_DEV.md#nix).
### only Frontend
This will use the backend of <https://app.windmill.dev> but your own frontend
with hot-code reloading. Note that you will need to use a username / password login due to CSRF checks using a different auth provider.
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO alerts (alert_type, message) VALUES ('recovered_critical_error', $1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "00ce4ed3ca0eac7cb6283b047353a64b9e78c4beb423f04baef9a53fbf87e9f9"
}
@@ -0,0 +1,48 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, alert_type, message, created_at, acknowledged \n FROM alerts \n WHERE acknowledged = $1\n ORDER BY created_at DESC \n LIMIT $2 OFFSET $3",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int4"
},
{
"ordinal": 1,
"name": "alert_type",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "message",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "acknowledged",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Bool",
"Int8",
"Int8"
]
},
"nullable": [
false,
false,
false,
false,
true
]
},
"hash": "0b955f2cff82a2d4ba3840588143e08952f029480d4a42503ecc3c5e70437995"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\" FROM http_trigger",
"query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\" FROM http_trigger",
"describe": {
"columns": [
{
@@ -65,6 +65,11 @@
}
}
}
},
{
"ordinal": 10,
"name": "static_asset_config: _",
"type_info": "Jsonb"
}
],
"parameters": {
@@ -80,8 +85,9 @@
false,
false,
false,
false
false,
true
]
},
"hash": "fe4f91ca7e179e58041a5c0b1a36015175e9301c01a99e8f2b82cb853349a183"
"hash": "11b698f82a54aac68b3617047dfe2b18dd6da7d962118fee276af354218baac2"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_flow->'failure_module' != 'null'::jsonb\n FROM completed_job\n WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "31dd23f6768052e486668172e15eba1a3f9b6a8e5678a7ee8e5456ff4405d6f9"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now())",
"query": "INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, static_asset_config, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now())",
"describe": {
"columns": [],
"parameters": {
@@ -27,11 +27,12 @@
}
}
},
"Jsonb",
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "c229744534f17f7622c3dee21bb1e7292ff17e6dffe58e80e53bff8baade07c8"
"hash": "333b484ffa030dee08e7b1161fcbc48af411377d2d9f58f92fc9d5eacdf0fba1"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO alerts (alert_type, message, acknowledged) VALUES ('recovered_critical_error', $1, $2)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "3a94ad52c6b7cde844fa868167248cd9ff63e5fdfa1d93d8fbec32a257b6b05e"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE http_trigger \n SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, edited_by = $7, email = $8, is_async = $9, requires_auth = $10, edited_at = now() \n WHERE workspace_id = $11 AND path = $12",
"query": "UPDATE http_trigger \n SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, static_asset_config = $7, edited_by = $8, email = $9, is_async = $10, requires_auth = $11, edited_at = now() \n WHERE workspace_id = $12 AND path = $13",
"describe": {
"columns": [],
"parameters": {
@@ -24,6 +24,7 @@
}
}
},
"Jsonb",
"Varchar",
"Varchar",
"Bool",
@@ -34,5 +35,5 @@
},
"nullable": []
},
"hash": "4d8640e84fccf1a0b799d8396a51e69345137e68d5096c70ba0a4332075d97ea"
"hash": "487d377e2df67fc3ea39d183ba9f99d45828d7c8e0ff10c5d74c454472e0493c"
}
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO alerts (alert_type, message) VALUES ('critical_error', $1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "4c0067c2135a259aea5cc2db60f7375a9a33671be8ef406427d90f67a98c9c9f"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO alerts (alert_type, message, acknowledged) VALUES ('critical_error', $1, $2)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "4d22084a5d9860832f30e8f08cbfa1848ed3c1336fa4790f45b8189c4ac97d91"
}
@@ -5,7 +5,7 @@
"columns": [
{
"ordinal": 0,
"name": "bool",
"name": "?column?",
"type_info": "Bool"
}
],
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, edited_by = $5, email = $6, is_async = $7, requires_auth = $8, edited_at = now() \n WHERE workspace_id = $9 AND path = $10",
"query": "UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, static_asset_config = $5, edited_by = $6, email = $7, is_async = $8, requires_auth = $9, edited_at = now() \n WHERE workspace_id = $10 AND path = $11",
"describe": {
"columns": [],
"parameters": {
@@ -22,6 +22,7 @@
}
}
},
"Jsonb",
"Varchar",
"Varchar",
"Bool",
@@ -32,5 +33,5 @@
},
"nullable": []
},
"hash": "55d44f569f8ebfccddf975e1a330ef0dc286f4138efe923832371cdbac7157b0"
"hash": "7113d7cc72e44e4b7e01b69cc18cbe7b0399cf8ec0e9e6d2b05ceef589c432df"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $11)\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14 FROM uuid_table) \n RETURNING id",
"query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $11)\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout, raw_code, raw_lock, raw_flow, flow_status)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14, $15, $16, $17, $18 FROM uuid_table) \n RETURNING id",
"describe": {
"columns": [
{
@@ -72,12 +72,16 @@
"Int4",
"Int4",
"Int4",
"Int4"
"Int4",
"Text",
"Text",
"Jsonb",
"Jsonb"
]
},
"nullable": [
false
]
},
"hash": "0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18"
"hash": "a69ba7471d1a5faa145bebbd17d43e6dbe02e9e92ebbc31a50c99c3a04719284"
}
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE alerts SET acknowledged = true WHERE acknowledged = false",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "a7c5008aa7ea43d0afac7d9f19846976ed7af2e90270902f001115e023cb947d"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_flow->'failure_module' != 'null'::jsonb\n FROM queue\n WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "aa6907b7266ee437dfbd77a9bf6c047fda88f3e72a0d10323a5e4020cf857c42"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT flow_version.value AS \"value: sqlx::types::Json<Box<RawValue>>\" FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 AND flow.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "value: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "ae543dfa106fa6ad4e9bf45cda1110d4702a80f455c63af6fcbd7cf45bbc4f7a"
}
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE alerts SET acknowledged = true WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int4"
]
},
"nullable": []
},
"hash": "be3ae231557e794172336bc27d725f862dcf039bbb3d75ced9d54c86f53d2580"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\" FROM http_trigger WHERE workspace_id = $1",
"query": "SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\" FROM http_trigger WHERE workspace_id = $1",
"describe": {
"columns": [
{
@@ -65,6 +65,11 @@
}
}
}
},
{
"ordinal": 10,
"name": "static_asset_config: _",
"type_info": "Jsonb"
}
],
"parameters": {
@@ -82,8 +87,9 @@
false,
false,
false,
false
false,
true
]
},
"hash": "02f1a6eeb27067dc438459238e7b016f5ccf9e3fe0ffbe88471f15aad8f74441"
"hash": "c9930fcfe79541af570eace58ba7e15a0816a6b4fd036cf7b991a210654b2633"
}
@@ -0,0 +1,47 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, alert_type, message, created_at, acknowledged \n FROM alerts \n ORDER BY created_at DESC \n LIMIT $1 OFFSET $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int4"
},
{
"ordinal": 1,
"name": "alert_type",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "message",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "acknowledged",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Int8",
"Int8"
]
},
"nullable": [
false,
false,
false,
false,
true
]
},
"hash": "cc5ab80241b88c5befea279f16c4ec68cec17b31dcd277b321f652917346496b"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as \"http_method: _\", edited_by, email, edited_at, extra_perms, is_async, requires_auth\n FROM http_trigger\n WHERE workspace_id = $1 AND path = $2",
"query": "SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as \"http_method: _\", edited_by, email, edited_at, extra_perms, is_async, requires_auth, static_asset_config as \"static_asset_config: _\"\n FROM http_trigger\n WHERE workspace_id = $1 AND path = $2",
"describe": {
"columns": [
{
@@ -80,6 +80,11 @@
"ordinal": 12,
"name": "requires_auth",
"type_info": "Bool"
},
{
"ordinal": 13,
"name": "static_asset_config: _",
"type_info": "Jsonb"
}
],
"parameters": {
@@ -101,8 +106,9 @@
false,
false,
false,
false
false,
true
]
},
"hash": "4fb95eae1c871241efe2ef79615ce03cba0e4a12aad3274e4829d98e38ca1491"
"hash": "f904702536c106b0e5da8facae119c6af887c49a29ae44b3a95350ff27fb1ccf"
}
+202 -185
View File
File diff suppressed because it is too large Load Diff
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "windmill"
version = "1.419.0"
version = "1.423.2"
authors.workspace = true
edition.workspace = true
@@ -29,7 +29,7 @@ members = [
]
[workspace.package]
version = "1.419.0"
version = "1.423.2"
authors = ["Ruben Fiszel <ruben@windmill.dev>"]
edition = "2021"
+1 -1
View File
@@ -1 +1 @@
aedaf655693cb9bab59fcf79a48c60aad6822668
e3df64af864f9d3546a97ff21f719c553e34ff1c
@@ -0,0 +1 @@
ALTER TABLE alerts DROP COLUMN acknowledged;
@@ -0,0 +1,9 @@
-- Step 1: Add the new column 'acknowledged' to the 'alerts' table
ALTER TABLE alerts
ADD COLUMN acknowledged BOOLEAN DEFAULT NULL;
-- Step 2: Update all existing rows to set 'acknowledged' to true
-- we don't want to pop up notifications to all users after migrations
-- but only show new alerts from the point of the upgrade
UPDATE alerts
SET acknowledged = TRUE;
@@ -0,0 +1,2 @@
-- Add down migration script here
ALTER TABLE http_trigger DROP COLUMN static_asset_config;
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE http_trigger ADD COLUMN static_asset_config JSONB;
@@ -62,6 +62,7 @@ static PYTHON_IMPORTS_REPLACEMENT: phf::Map<&'static str, &'static str> = phf_ma
"lokalise" => "python-lokalise-api",
"msgraph" => "msgraph-sdk",
"pythonjsonlogger" => "python-json-logger",
"socks" => "PySocks",
};
fn replace_import(x: String) -> String {
+8 -2
View File
@@ -36,7 +36,7 @@ use windmill_common::{
ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING,
OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
},
@@ -78,7 +78,7 @@ use crate::monitor::{
reload_hub_base_url_setting, reload_job_default_timeout_setting, reload_jwt_secret_setting,
reload_license_key, reload_npm_config_registry_setting, reload_pip_index_url_setting,
reload_retention_period_setting, reload_scim_token_setting, reload_smtp_config,
reload_worker_config,
reload_worker_config, reload_critical_alert_mute_ui_setting,
};
#[cfg(feature = "parquet")]
@@ -817,6 +817,12 @@ Windmill Community Edition {GIT_VERSION}
tracing::error!(error = %e, "Could not reload jwt secret setting");
}
},
CRITICAL_ALERT_MUTE_UI_SETTING => {
tracing::info!("Critical alert UI setting changed");
if let Err(e) = reload_critical_alert_mute_ui_setting(&db).await {
tracing::error!(error = %e, "Could not reload critical alert UI setting");
}
},
a @_ => {
tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a);
}
+17 -2
View File
@@ -39,7 +39,7 @@ use windmill_common::{
EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING,
KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING,
PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
},
@@ -55,7 +55,7 @@ use windmill_common::{
WORKER_GROUP,
},
BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
METRICS_DEBUG_ENABLED, METRICS_ENABLED,
METRICS_DEBUG_ENABLED, METRICS_ENABLED, CRITICAL_ALERT_MUTE_UI_ENABLED
};
use windmill_queue::cancel_job;
use windmill_worker::{
@@ -131,6 +131,10 @@ pub async fn initial_load(
tracing::error!("Error loading expose debug metrics: {e:#}");
}
if let Err(e) = reload_critical_alert_mute_ui_setting(db).await {
tracing::error!("Error loading critical alert mute ui setting: {e:#}");
}
if let Err(e) = load_tag_per_workspace_enabled(db).await {
tracing::error!("Error loading default tag per workpsace: {e:#}");
}
@@ -226,6 +230,17 @@ pub async fn load_tag_per_workspace_workspaces(db: &DB) -> error::Result<()> {
Ok(())
}
pub async fn reload_critical_alert_mute_ui_setting(db: &DB) -> error::Result<()> {
let mute = load_value_from_global_settings(db, CRITICAL_ALERT_MUTE_UI_SETTING).await;
match mute {
Ok(Some(serde_json::Value::Bool(t))) => {
CRITICAL_ALERT_MUTE_UI_ENABLED.store(t, Ordering::Relaxed);
}
_ => (),
};
Ok(())
}
pub async fn load_metrics_debug_enabled(db: &DB) -> error::Result<()> {
let metrics_enabled = load_value_from_global_settings(db, EXPOSE_DEBUG_METRICS_SETTING).await;
match metrics_enabled {
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+139 -2
View File
@@ -1,7 +1,7 @@
openapi: "3.0.3"
info:
version: 1.419.0
version: 1.423.2
title: Windmill API
contact:
@@ -190,7 +190,7 @@ paths:
schema:
type: string
/w/{workspace}/users/{username}:
/w/{workspace}/users/get/{username}:
get:
summary: get user (require admin privilege)
operationId: getUser
@@ -773,6 +773,78 @@ paths:
schema:
type: string
/settings/critical_alerts:
get:
summary: Get all critical alerts
operationId: getCriticalAlerts
tags:
- setting
parameters:
- in: query
name: page
schema:
type: integer
default: 1
description: The page number to retrieve (minimum value is 1)
- in: query
name: page_size
schema:
type: integer
default: 10
maximum: 100
description: Number of alerts per page (maximum is 100)
- in: query
name: acknowledged
schema:
type: boolean
nullable: true
description: Filter by acknowledgment status; true for acknowledged, false for unacknowledged, and omit for all alerts
responses:
"200":
description: Successfully retrieved all critical alerts
content:
application/json:
schema:
type: array
items:
$ref: '#/components/schemas/CriticalAlert'
/settings/critical_alerts/{id}/acknowledge:
post:
summary: Acknowledge a critical alert
operationId: acknowledgeCriticalAlert
tags:
- setting
parameters:
- in: path
name: id
required: true
schema:
type: integer
description: The ID of the critical alert to acknowledge
responses:
"200":
description: Successfully acknowledged the critical alert
content:
application/json:
schema:
type: string
example: "Critical alert acknowledged"
/settings/critical_alerts/acknowledge_all:
post:
summary: Acknowledge all unacknowledged critical alerts
operationId: acknowledgeAllCriticalAlerts
tags:
- setting
responses:
"200":
description: Successfully acknowledged all unacknowledged critical alerts.
content:
application/json:
schema:
type: string
example: "All unacknowledged critical alerts acknowledged"
/settings/test_license_key:
post:
@@ -9146,6 +9218,14 @@ paths:
in: query
schema:
type: string
- name: content_type
in: query
schema:
type: string
- name: content_disposition
in: query
schema:
type: string
requestBody:
description: File content
required: true
@@ -11463,6 +11543,17 @@ components:
type: string
route_path:
type: string
static_asset_config:
type: object
properties:
s3:
type: string
storage:
type: string
filename:
type: string
required:
- s3
is_flow:
type: boolean
extra_perms:
@@ -11509,6 +11600,17 @@ components:
type: string
route_path:
type: string
static_asset_config:
type: object
properties:
s3:
type: string
storage:
type: string
filename:
type: string
required:
- s3
is_flow:
type: boolean
http_method:
@@ -11542,6 +11644,17 @@ components:
type: string
route_path:
type: string
static_asset_config:
type: object
properties:
s3:
type: string
storage:
type: string
filename:
type: string
required:
- s3
is_flow:
type: boolean
http_method:
@@ -12073,6 +12186,10 @@ components:
type: object
additionalProperties:
type: object
s3_inputs:
type: array
items:
type: object
execution_mode:
type: string
enum: [viewer, publisher, anonymous]
@@ -12680,3 +12797,23 @@ components:
type: string
format: date-time
CriticalAlert:
type: object
properties:
id:
type: integer
description: Unique identifier for the alert
alert_type:
type: string
description: Type of alert (e.g., critical_error)
message:
type: string
description: The message content of the alert
created_at:
type: string
format: date-time
description: Time when the alert was created
acknowledged:
type: boolean
nullable: true
description: Acknowledgment status of the alert, can be true, false, or null if not set
+312 -38
View File
@@ -7,6 +7,12 @@ use std::collections::HashMap;
* Please see the included NOTICE for copyright information and
* LICENSE-AGPL for a copy of the license.
*/
#[cfg(feature = "parquet")]
use crate::{job_helpers_ee::{
get_random_file_name, get_s3_resource, get_workspace_s3_resource, upload_file_internal,
UploadFileResponse,
}, users::fetch_api_authed_from_permissioned_as};
use crate::{
db::{ApiAuthed, DB},
resources::get_resource_value_interpolated_internal,
@@ -23,7 +29,13 @@ use axum::{
Router,
};
use hyper::StatusCode;
#[cfg(feature = "parquet")]
use itertools::Itertools;
use magic_crypt::MagicCryptTrait;
#[cfg(feature = "parquet")]
use object_store::{Attribute, Attributes};
#[cfg(feature = "parquet")]
use regex::Regex;
use serde::{Deserialize, Serialize};
use serde_json::{json, value::RawValue};
use sha2::{Digest, Sha256};
@@ -32,6 +44,8 @@ use sqlx::{types::Uuid, FromRow};
use std::str;
use windmill_audit::audit_ee::audit_log;
use windmill_audit::ActionKind;
#[cfg(feature = "parquet")]
use windmill_common::s3_helpers::build_object_store_client;
use windmill_common::{
apps::ListAppQuery,
db::UserDB,
@@ -69,6 +83,7 @@ pub fn workspaced_service() -> Router {
pub fn unauthed_service() -> Router {
Router::new()
.route("/execute_component/*path", post(execute_component))
.route("/upload_s3_file/*path", post(upload_s3_file_from_app))
.route("/public_app/:secret", get(get_public_app_by_secret))
.route("/public_resource/*path", get(get_public_resource))
}
@@ -179,6 +194,14 @@ pub struct PolicyTriggerableInputs {
allow_user_resources: AllowUserResources,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct S3Input {
allowed_resources: Vec<String>,
allow_user_resources: bool,
allow_workspace_resource: bool,
file_key_regex: String,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Policy {
pub on_behalf_of: Option<String>,
@@ -192,6 +215,7 @@ pub struct Policy {
#[serde(skip_serializing_if = "Option::is_none")]
pub triggerables_v2: Option<HashMap<String, PolicyTriggerableInputs>>,
pub execution_mode: ExecutionMode,
pub s3_inputs: Option<Vec<S3Input>>,
}
#[derive(Deserialize)]
@@ -432,9 +456,7 @@ async fn get_latest_version(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<Option<AppHistory>> {
let mut tx = user_db.begin(&authed).await?;
let row = sqlx::query!(
"SELECT a.id as app_id, av.id as version_id, dm.deployment_msg as deployment_msg
@@ -457,7 +479,6 @@ async fn get_latest_version(
} else {
return Ok(Json(None));
}
}
async fn update_app_history(
@@ -1067,6 +1088,49 @@ fn digest(code: &str) -> String {
format!("rawscript/{:x}", result)
}
async fn get_on_behalf_details_from_policy_and_authed(
policy: &Policy,
opt_authed: &Option<ApiAuthed>,
) -> Result<(String, String, String)> {
let (username, permissioned_as, email) = match policy.execution_mode {
ExecutionMode::Anonymous => {
let username = opt_authed
.as_ref()
.map(|a| a.username.clone())
.unwrap_or_else(|| "anonymous".to_string());
let (permissioned_as, email) = get_on_behalf_of(&policy)?;
(username, permissioned_as, email)
}
ExecutionMode::Publisher => {
let username = opt_authed
.as_ref()
.map(|a| a.username.clone())
.ok_or_else(|| {
Error::BadRequest(
"publisher execution mode requires authentication".to_string(),
)
})?;
let (permissioned_as, email) = get_on_behalf_of(&policy)?;
(username, permissioned_as, email)
}
ExecutionMode::Viewer => {
let (username, email) = opt_authed
.as_ref()
.map(|a| (a.username.clone(), a.email.clone()))
.ok_or_else(|| {
Error::BadRequest("Required to be authed in viewer mode".to_string())
})?;
(
username.clone(),
username_to_permissioned_as(&username),
email,
)
}
};
Ok((username, permissioned_as, email))
}
async fn execute_component(
OptAuthed(opt_authed): OptAuthed,
Extension(db): Extension<DB>,
@@ -1129,6 +1193,7 @@ async fn execute_component(
triggerables_v2: Some(hm),
on_behalf_of: None,
on_behalf_of_email: None,
s3_inputs: None,
}
}
_ => {
@@ -1146,41 +1211,8 @@ async fn execute_component(
}
};
let (username, permissioned_as, email) = match policy.execution_mode {
ExecutionMode::Anonymous => {
let username = opt_authed
.as_ref()
.map(|a| a.username.clone())
.unwrap_or_else(|| "anonymous".to_string());
let (permissioned_as, email) = get_on_behalf_of(&policy)?;
(username, permissioned_as, email)
}
ExecutionMode::Publisher => {
let username = opt_authed
.as_ref()
.map(|a| a.username.clone())
.ok_or_else(|| {
Error::BadRequest(
"publisher execution mode requires authentication".to_string(),
)
})?;
let (permissioned_as, email) = get_on_behalf_of(&policy)?;
(username, permissioned_as, email)
}
ExecutionMode::Viewer => {
let (username, email) = opt_authed
.as_ref()
.map(|a| (a.username.clone(), a.email.clone()))
.ok_or_else(|| {
Error::BadRequest("Required to be authed in viewer mode".to_string())
})?;
(
username.clone(),
username_to_permissioned_as(&username),
email,
)
}
};
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, .. } => {
@@ -1249,6 +1281,248 @@ async fn execute_component(
Ok(uuid.to_string())
}
#[cfg(not(feature = "parquet"))]
async fn upload_s3_file_from_app() -> Result<()> {
return Err(Error::BadRequest(
"This endpoint requires the parquet feature to be enabled".to_string(),
));
}
#[cfg(feature = "parquet")]
#[derive(Debug, Deserialize, Clone)]
struct UploadFileToS3Query {
file_key: Option<String>,
file_extension: Option<String>,
s3_resource_path: Option<String>,
content_type: Option<String>,
content_disposition: Option<String>,
force_viewer_file_key_regex: Option<String>,
force_viewer_allow_user_resources: Option<bool>,
force_viewer_allow_workspace_resource: Option<bool>,
force_viewer_allowed_resources: Option<String>,
}
#[cfg(feature = "parquet")]
async fn upload_s3_file_from_app(
OptAuthed(opt_authed): OptAuthed,
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
Query(query): Query<UploadFileToS3Query>,
request: axum::extract::Request,
) -> JsonResult<UploadFileResponse> {
let policy = if let Some(file_key_regex) = query.force_viewer_file_key_regex {
Some(Policy {
execution_mode: ExecutionMode::Viewer,
triggerables: None,
triggerables_v2: None,
on_behalf_of: None,
on_behalf_of_email: None,
s3_inputs: Some(vec![S3Input {
file_key_regex: file_key_regex,
allow_user_resources: query.force_viewer_allow_user_resources.unwrap_or(false),
allow_workspace_resource: query
.force_viewer_allow_workspace_resource
.unwrap_or(false),
allowed_resources: query
.force_viewer_allowed_resources
.map(|s| s.split(',').map(|s| s.to_string()).collect())
.unwrap_or_default(),
}]),
})
} else {
let policy_o = sqlx::query_scalar!(
"SELECT policy from app WHERE path = $1 AND workspace_id = $2",
&path.0,
&w_id
)
.fetch_optional(&db)
.await?;
policy_o
.map(|p| serde_json::from_value::<Policy>(p).map_err(to_anyhow))
.transpose()?
};
let user_db = UserDB::new(db.clone());
let (s3_resource_opt, file_key) = if policy.as_ref().is_some_and(|p| p.s3_inputs.is_some()) {
let policy = policy.unwrap();
let s3_inputs = policy.s3_inputs.as_ref().unwrap();
let (username, permissioned_as, email) =
get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?;
let on_behalf_authed =
fetch_api_authed_from_permissioned_as(permissioned_as, email, &w_id, &db, username)
.await?;
if let Some(file_key) = query.file_key {
// file key is provided => requires workspace, user or list policy and must match the regex
let matching_s3_inputs = if let Some(ref s3_resource_path) = query.s3_resource_path {
s3_inputs
.iter()
.filter(|s3_input| {
s3_input.allowed_resources.contains(s3_resource_path)
|| s3_input.allow_user_resources
})
.sorted_by_key(|i| i.allow_user_resources) // consider user resources last
.collect::<Vec<_>>()
} else {
s3_inputs
.iter()
.filter(|s3_input| s3_input.allow_workspace_resource)
.collect::<Vec<_>>()
};
let matched_input = matching_s3_inputs.iter().find(|s3_input| {
match Regex::new(&s3_input.file_key_regex) {
Ok(re) => re.is_match(&file_key),
Err(e) => {
tracing::error!("Error compiling regex: {}", e);
false
}
}
});
if let Some(matched_input) = matched_input {
if let Some(ref s3_resource_path) = query.s3_resource_path {
if matched_input.allow_user_resources {
if let Some(authed) = opt_authed {
(
Some(
get_s3_resource(
&authed,
&db,
Some(user_db),
"",
&w_id,
s3_resource_path,
None,
None,
)
.await?,
),
file_key,
)
} else {
return Err(Error::BadRequest(
"User resources are not allowed without being logged in"
.to_string(),
));
}
} else {
(
Some(
get_s3_resource(
&on_behalf_authed,
&db,
Some(user_db),
"",
&w_id,
s3_resource_path,
None,
None,
)
.await?,
),
file_key,
)
}
} else {
let (_, s3_resource_opt) =
get_workspace_s3_resource(&on_behalf_authed, &db, None, "", &w_id, None)
.await?;
(s3_resource_opt, file_key)
}
} else {
return Err(Error::BadRequest(
"No matching s3 resource found for the given file key".to_string(),
));
}
} else {
// no file key => requires unnamed upload policy => allow workspace resource and file_key_regex is empty
let has_unnamed_policy = s3_inputs.iter().any(|s3_input| {
s3_input.allow_workspace_resource && s3_input.file_key_regex.is_empty()
});
if !has_unnamed_policy {
return Err(Error::BadRequest(
"no policy found for unnamed s3 file uplooad".to_string(),
));
}
// for now, we place all files into `windmill_uploads` folder with a random name
// TODO: make the folder configurable via the workspace settings
let file_key = get_random_file_name(query.file_extension);
let (_, s3_resource_opt) =
get_workspace_s3_resource(&on_behalf_authed, &db, None, "", &w_id, None).await?;
(s3_resource_opt, file_key)
}
} else {
// backward compatibility (no policy)
// if no policy but logged in, use the user's auth to get the s3 resource
if let Some(authed) = opt_authed {
let file_key = query
.file_key
.unwrap_or_else(|| get_random_file_name(query.file_extension));
if let Some(ref s3_resource_path) = query.s3_resource_path {
(
Some(
get_s3_resource(
&authed,
&db,
Some(user_db),
"",
&w_id,
s3_resource_path,
None,
None,
)
.await?,
),
file_key,
)
} else {
let (_, s3_resource) =
get_workspace_s3_resource(&authed, &db, None, "", &w_id, None).await?;
(s3_resource, file_key)
}
} else {
return Err(Error::BadRequest("Missing s3 policy".to_string()));
}
};
let s3_resource = s3_resource_opt.ok_or(Error::InternalErr(
"No files storage resource defined at the workspace level".to_string(),
))?;
let s3_client = build_object_store_client(&s3_resource).await?;
let options = Attributes::from_iter(vec![
(
Attribute::ContentType,
query.content_type.unwrap_or_else(|| {
mime_guess::from_path(&file_key)
.first_or_octet_stream()
.to_string()
}),
),
(
Attribute::ContentDisposition,
query.content_disposition.unwrap_or("inline".to_string()),
),
])
.into();
upload_file_internal(s3_client, &file_key, request, options).await?;
return Ok(Json(UploadFileResponse { file_key }));
}
fn get_on_behalf_of(policy: &Policy) -> Result<(String, String)> {
let permissioned_as = policy
.on_behalf_of
+129 -32
View File
@@ -1,24 +1,5 @@
use axum::{
extract::{Path, Query},
response::IntoResponse,
routing::{delete, get, post},
Extension, Json, Router,
};
use http::{HeaderMap, StatusCode};
use serde::{Deserialize, Serialize};
use sql_builder::{bind::Bind, SqlBuilder};
use sqlx::prelude::FromRow;
use std::collections::HashMap;
use tower_http::cors::CorsLayer;
use windmill_audit::{audit_ee::audit_log, ActionKind};
use windmill_common::{
db::UserDB,
error::{self, JsonResult},
utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
worker::{to_raw_value, CLOUD_HOSTED},
};
use windmill_queue::PushArgsOwned;
#[cfg(feature = "parquet")]
use crate::job_helpers_ee::get_workspace_s3_resource;
use crate::{
db::{ApiAuthed, DB},
jobs::{
@@ -27,6 +8,31 @@ use crate::{
},
users::{fetch_api_authed, OptAuthed},
};
use axum::{
extract::{Path, Query},
response::IntoResponse,
routing::{delete, get, post},
Extension, Json, Router,
};
#[cfg(feature = "parquet")]
use http::header::IF_NONE_MATCH;
use http::{HeaderMap, StatusCode};
use serde::{Deserialize, Serialize};
use sql_builder::{bind::Bind, SqlBuilder};
use sqlx::prelude::FromRow;
use std::collections::HashMap;
use tower_http::cors::CorsLayer;
use windmill_audit::{audit_ee::audit_log, ActionKind};
#[cfg(feature = "parquet")]
use windmill_common::s3_helpers::build_object_store_client;
use windmill_common::{
db::UserDB,
error::{self, JsonResult},
s3_helpers::S3Object,
utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
worker::{to_raw_value, CLOUD_HOSTED},
};
use windmill_queue::PushArgsOwned;
lazy_static::lazy_static! {
static ref ROUTE_PATH_KEY_RE: regex::Regex = regex::Regex::new(r"/:\w+").unwrap();
@@ -99,6 +105,7 @@ struct NewTrigger {
is_async: bool,
requires_auth: bool,
http_method: HttpMethod,
static_asset_config: Option<sqlx::types::Json<S3Object>>,
}
#[derive(FromRow, Serialize)]
@@ -116,6 +123,7 @@ struct Trigger {
is_async: bool,
requires_auth: bool,
http_method: HttpMethod,
static_asset_config: Option<sqlx::types::Json<S3Object>>,
}
#[derive(Deserialize)]
@@ -127,6 +135,7 @@ struct EditTrigger {
is_async: bool,
requires_auth: bool,
http_method: HttpMethod,
static_asset_config: Option<sqlx::types::Json<S3Object>>,
}
#[derive(Deserialize)]
@@ -182,7 +191,7 @@ async fn get_trigger(
let path = path.to_path();
let trigger = sqlx::query_as!(
Trigger,
r#"SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as "http_method: _", edited_by, email, edited_at, extra_perms, is_async, requires_auth
r#"SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, http_method as "http_method: _", edited_by, email, edited_at, extra_perms, is_async, requires_auth, static_asset_config as "static_asset_config: _"
FROM http_trigger
WHERE workspace_id = $1 AND path = $2"#,
w_id,
@@ -209,7 +218,7 @@ async fn create_trigger(
let mut tx = user_db.begin(&authed).await?;
sqlx::query!(
"INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now())",
"INSERT INTO http_trigger (workspace_id, path, route_path, route_path_key, script_path, is_flow, is_async, requires_auth, http_method, static_asset_config, edited_by, email, edited_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now())",
w_id,
ct.path,
ct.route_path,
@@ -218,7 +227,8 @@ async fn create_trigger(
ct.is_flow,
ct.is_async,
ct.requires_auth,
ct.http_method as HttpMethod,
ct.http_method as _,
ct.static_asset_config as _,
&authed.username,
&authed.email
)
@@ -261,14 +271,15 @@ async fn update_trigger(
sqlx::query!(
"UPDATE http_trigger
SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, edited_by = $7, email = $8, is_async = $9, requires_auth = $10, edited_at = now()
WHERE workspace_id = $11 AND path = $12",
SET route_path = $1, route_path_key = $2, script_path = $3, path = $4, is_flow = $5, http_method = $6, static_asset_config = $7, edited_by = $8, email = $9, is_async = $10, requires_auth = $11, edited_at = now()
WHERE workspace_id = $12 AND path = $13",
ct.route_path,
&route_path_key,
ct.script_path,
ct.path,
ct.is_flow,
ct.http_method as HttpMethod,
ct.http_method as _,
ct.static_asset_config as _,
&authed.username,
&authed.email,
ct.is_async,
@@ -279,12 +290,13 @@ async fn update_trigger(
.execute(&mut *tx).await?;
} else {
sqlx::query!(
"UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, edited_by = $5, email = $6, is_async = $7, requires_auth = $8, edited_at = now()
WHERE workspace_id = $9 AND path = $10",
"UPDATE http_trigger SET script_path = $1, path = $2, is_flow = $3, http_method = $4, static_asset_config = $5, edited_by = $6, email = $7, is_async = $8, requires_auth = $9, edited_at = now()
WHERE workspace_id = $10 AND path = $11",
ct.script_path,
ct.path,
ct.is_flow,
ct.http_method as HttpMethod,
ct.http_method as _,
ct.static_asset_config as _,
&authed.username,
&authed.email,
ct.is_async,
@@ -405,6 +417,7 @@ struct TriggerRoute {
edited_by: String,
email: String,
http_method: HttpMethod,
static_asset_config: Option<sqlx::types::Json<S3Object>>,
}
async fn get_http_route_trigger(
@@ -421,7 +434,7 @@ async fn get_http_route_trigger(
let route_path = StripPath(splitted.collect::<Vec<_>>().join("/"));
let triggers = sqlx::query_as!(
TriggerRoute,
r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _" FROM http_trigger WHERE workspace_id = $1"#,
r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _", static_asset_config as "static_asset_config: _" FROM http_trigger WHERE workspace_id = $1"#,
w_id
)
.fetch_all(db)
@@ -430,7 +443,7 @@ async fn get_http_route_trigger(
} else {
let triggers = sqlx::query_as!(
TriggerRoute,
r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _" FROM http_trigger"#,
r#"SELECT path, script_path, is_flow, route_path, workspace_id, is_async, requires_auth, edited_by, email, http_method as "http_method: _", static_asset_config as "static_asset_config: _" FROM http_trigger"#,
)
.fetch_all(db)
.await?;
@@ -518,6 +531,90 @@ async fn route_job(
Ok(trigger) => trigger,
Err(e) => return e.into_response(),
};
#[cfg(not(feature = "parquet"))]
if trigger.static_asset_config.is_some() {
return error::Error::InternalErr(
"Static asset configuration is not supported in this build".to_string(),
)
.into_response();
}
#[cfg(feature = "parquet")]
if let Some(sqlx::types::Json(config)) = trigger.static_asset_config {
let build_static_response_f = async {
let (_, s3_resource_opt) = get_workspace_s3_resource(
&authed,
&db,
None,
&"NO_TOKEN".to_string(), // no token is provided in this case
&trigger.workspace_id,
config.storage,
)
.await?;
let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
"No files storage resource defined at the workspace level".to_string(),
))?;
let s3_client = build_object_store_client(&s3_resource).await?;
let path = object_store::path::Path::from(config.s3);
let s3_object = s3_client.get(&path).await.map_err(|err| {
tracing::warn!("Error retrieving file from S3: {:?}", err);
error::Error::InternalErr(format!("Error retrieving file: {}", err.to_string()))
})?;
let mut response_headers = http::HeaderMap::new();
if let Some(ref e_tag) = s3_object.meta.e_tag {
if let Some(if_none_match) = headers.get(IF_NONE_MATCH) {
if if_none_match == e_tag {
return Ok::<_, error::Error>((
StatusCode::NOT_MODIFIED,
response_headers,
axum::body::Body::empty(),
));
}
}
if let Ok(e_tag) = e_tag.parse() {
response_headers.insert("etag", e_tag);
}
}
response_headers.insert(
"content-type",
s3_object
.attributes
.get(&object_store::Attribute::ContentType)
.map(|s| s.parse().ok())
.flatten()
.unwrap_or("application/octet-stream".parse().unwrap()),
);
response_headers.insert(
"content-disposition",
config.filename.as_ref().map_or_else(
|| {
s3_object
.attributes
.get(&object_store::Attribute::ContentDisposition)
.map(|s| s.parse().ok())
.flatten()
.unwrap_or("inline".parse().unwrap())
},
|filename| {
format!("inline; filename=\"{}\"", filename)
.parse()
.unwrap_or("inline".parse().unwrap())
},
),
);
let body_stream = axum::body::Body::from_stream(s3_object.into_stream());
Ok::<_, error::Error>((StatusCode::OK, response_headers, body_stream))
};
match build_static_response_f.await {
Ok((status, headers, body_stream)) => {
return (status, headers, body_stream).into_response()
}
Err(e) => return e.into_response(),
}
}
let headers = headers
.iter()
.map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
@@ -1,5 +1,61 @@
use axum::Router;
use serde::Serialize;
use uuid::Uuid;
use windmill_common::s3_helpers::StorageResourceType;
#[cfg(feature = "parquet")]
use crate::db::{ApiAuthed, DB};
#[cfg(feature = "parquet")]
use object_store::{ObjectStore, PutMultipartOpts};
#[cfg(feature = "parquet")]
use std::sync::Arc;
use windmill_common::error;
#[cfg(feature = "parquet")]
use windmill_common::{db::UserDB, s3_helpers::ObjectStoreResource};
#[derive(Serialize)]
pub struct UploadFileResponse {
pub file_key: String,
}
pub fn workspaced_service() -> Router {
Router::new()
}
#[cfg(feature = "parquet")]
pub async fn get_workspace_s3_resource<'c>(
_authed: &ApiAuthed,
_db: &DB,
_user_db: Option<UserDB>,
_token: &str,
_w_id: &str,
_storage: Option<String>,
) -> windmill_common::error::Result<(Option<bool>, Option<ObjectStoreResource>)> {
// implementation is not open source
Ok((None, None))
}
pub fn get_random_file_name(_file_extension: Option<String>) -> String {
todo!()
}
pub async fn get_s3_resource<'c>(
_authed: &ApiAuthed,
_db: &DB,
_user_db: Option<UserDB>,
_token: &str,
_w_id: &str,
_resource_path: &str,
_resource_type: Option<StorageResourceType>,
_job_id: Option<Uuid>,
) -> error::Result<ObjectStoreResource> {
todo!()
}
pub async fn upload_file_internal(
_s3_client: Arc<dyn ObjectStore>,
_file_key: &str,
_request: axum::extract::Request,
_options: PutMultipartOpts,
) -> error::Result<()> {
todo!()
}
+86 -51
View File
@@ -65,7 +65,7 @@ use windmill_common::{
db::UserDB,
error::{self, to_anyhow, Error},
flow_status::{Approval, FlowStatus, FlowStatusModule},
flows::FlowValue,
flows::{add_virtual_items_if_necessary, FlowValue},
jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode},
oauth2::HmacSha256,
scripts::{ScriptHash, ScriptLang},
@@ -4275,18 +4275,26 @@ async fn run_flow_dependencies_job(
wait_result
}
#[derive(Deserialize)]
struct BatchRawScript {
content: String,
language: Option<ScriptLang>,
lock: Option<String>,
}
#[derive(Deserialize)]
struct BatchInfo {
kind: String,
flow_value: Option<FlowValue>,
path: Option<String>,
rawscript: Option<BatchRawScript>,
}
#[tracing::instrument(level = "trace", skip_all)]
async fn add_batch_jobs(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
Extension(_rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
Path((w_id, n)): Path<(String, i32)>,
Json(batch_info): Json<BatchInfo>,
) -> error::JsonResult<Vec<Uuid>> {
@@ -4302,6 +4310,10 @@ async fn add_batch_jobs(
concurrent_limit,
concurrent_time_window_s,
timeout,
raw_code,
raw_lock,
raw_flow,
flow_status
) = match batch_info.kind.as_str() {
"script" => {
if let Some(path) = batch_info.path {
@@ -4329,6 +4341,10 @@ async fn add_batch_jobs(
concurrent_limit,
concurrency_time_window_s,
timeout,
None,
None,
None,
None,
)
} else {
Err(anyhow::anyhow!(
@@ -4336,61 +4352,72 @@ async fn add_batch_jobs(
))?
}
}
"flow" => {
let mut uuids: Vec<Uuid> = Vec::new();
let payload = if let Some(ref fv) = batch_info.flow_value {
JobPayload::RawFlow { value: fv.clone(), path: None, restarted_from: None }
} else {
if let Some(path) = batch_info.path.as_ref() {
JobPayload::Flow {
path: path.to_string(),
dedicated_worker: None,
apply_preprocessor: false,
}
} else {
Err(anyhow::anyhow!(
"Path is required if no value is not provided"
))?
}
};
let mut tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq);
for _ in 0..n {
let ehm = HashMap::new();
let (uuid, ntx) = push(
&db,
tx,
&w_id,
payload.clone(),
PushArgs::from(&ehm),
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
"rawscript" => {
if let Some(rawscript) = batch_info.rawscript {
(
None,
None,
JobKind::Preview,
rawscript.language,
None,
None,
None,
None,
None,
false,
false,
None,
true,
Some(rawscript.content),
rawscript.lock,
None,
None,
None,
None,
Some(&authed.clone().into()),
)
.await?;
tx = PushIsolationLevel::Transaction(ntx);
uuids.push(uuid);
} else {
Err(anyhow::anyhow!(
"rawscript is required for `rawscript` kind"
))?
}
match tx {
PushIsolationLevel::Transaction(tx) => {
tx.commit().await?;
}
_ => (),
}
return Ok(Json(uuids));
}
"flow" => {
let (mut value, job_kind, path) = if let Some(value) = batch_info.flow_value {
(value, JobKind::FlowPreview, None)
} else if let Some(path) = batch_info.path {
let value_json = sqlx::query!(
"SELECT flow_version.value AS \"value: sqlx::types::Json<Box<RawValue>>\" FROM flow
LEFT JOIN flow_version
ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.path = $1 AND flow.workspace_id = $2",
&path, &w_id
)
.fetch_optional(&db)
.await?
.ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?;
let value =
serde_json::from_str::<FlowValue>(value_json.value.get()).map_err(|err| {
Error::InternalErr(format!(
"could not convert json to flow for {path}: {err:?}"
))
})?;
(value, JobKind::Flow, Some(path))
} else {
Err(anyhow::anyhow!(
"Path is required if no value is not provided"
))?
};
add_virtual_items_if_necessary(&mut value.modules);
let flow_status = FlowStatus::new(&value);
(
None, // script_hash
path, // script_path
job_kind, // job_kind
None, // language
None, // dedicated_worker
value.concurrency_key.clone(), // custom_concurrency_key
value.concurrent_limit.clone(), // concurrent_limit
value.concurrency_time_window_s, // concurrency_time_window_s
None, // timeout
None, // raw_code
None, // raw_lock
Some(value), // raw_flow
Some(flow_status), // flow_status
)
}
"noop" => (
None,
@@ -4402,6 +4429,10 @@ async fn add_batch_jobs(
None,
None,
None,
None,
None,
None,
None,
),
_ => {
return Err(error::Error::BadRequest(format!(
@@ -4428,8 +4459,8 @@ async fn add_batch_jobs(
select gen_random_uuid() as uuid from generate_series(1, $11)
)
INSERT INTO queue
(id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout)
(SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14 FROM uuid_table)
(id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout, raw_code, raw_lock, raw_flow, flow_status)
(SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14, $15, $16, $17, $18 FROM uuid_table)
RETURNING id"#,
hash.map(|h| h.0),
path,
@@ -4444,7 +4475,11 @@ async fn add_batch_jobs(
n,
concurrent_limit,
concurrent_time_window_s,
timeout
timeout,
raw_code,
raw_lock,
raw_flow.map(sqlx::types::Json) as Option<sqlx::types::Json<FlowValue>>,
flow_status.map(sqlx::types::Json) as Option<sqlx::types::Json<FlowStatus>>
)
.fetch_all(&db)
.await?;
+118 -1
View File
@@ -58,7 +58,10 @@ pub fn global_service() -> Router {
)
.route("/renew_license_key", post(renew_license_key))
.route("/customer_portal", post(create_customer_portal_session))
.route("/test_critical_channels", post(test_critical_channels));
.route("/test_critical_channels", post(test_critical_channels))
.route("/critical_alerts", get(get_critical_alerts))
.route("/critical_alerts/:id/acknowledge", post(acknowledge_critical_alert))
.route("/critical_alerts/acknowledge_all", post(acknowledge_all_critical_alerts));
#[cfg(feature = "parquet")]
{
@@ -430,3 +433,117 @@ pub async fn test_critical_channels(
pub async fn test_critical_channels() -> Result<String> {
Ok("Critical channels require EE".to_string())
}
use serde::Serialize;
#[derive(Serialize)]
pub struct CriticalAlert {
id: i32,
alert_type: String,
message: String,
created_at: chrono::DateTime<chrono::Utc>,
acknowledged: Option<bool>,
}
#[cfg(feature = "enterprise")]
#[derive(Deserialize)]
pub struct AlertQueryParams {
pub page: Option<i32>,
pub page_size: Option<i32>,
pub acknowledged: Option<bool>,
}
#[cfg(feature = "enterprise")]
pub async fn get_critical_alerts(
Extension(db): Extension<DB>,
authed: ApiAuthed,
Query(params): Query<AlertQueryParams>,
) -> JsonResult<Vec<CriticalAlert>> {
require_super_admin(&db, &authed.email).await?;
// Default pagination values if not provided
let page = params.page.unwrap_or(1).max(1);
let page_size = params.page_size.unwrap_or(10).min(100) as i64;
let offset = ((page - 1) * page_size as i32) as i64;
let alerts = if let Some(acknowledged) = params.acknowledged {
sqlx::query_as!(
CriticalAlert,
"SELECT id, alert_type, message, created_at, acknowledged
FROM alerts
WHERE acknowledged = $1
ORDER BY created_at DESC
LIMIT $2 OFFSET $3",
acknowledged,
page_size,
offset
)
.fetch_all(&db)
.await?
} else {
sqlx::query_as!(
CriticalAlert,
"SELECT id, alert_type, message, created_at, acknowledged
FROM alerts
ORDER BY created_at DESC
LIMIT $1 OFFSET $2",
page_size,
offset
)
.fetch_all(&db)
.await?
};
Ok(Json(alerts))
}
#[cfg(not(feature = "enterprise"))]
pub async fn get_critical_alerts() -> error::Error {
error::Error::NotFound("Critical Alerts require EE".to_string())
}
#[cfg(feature = "enterprise")]
pub async fn acknowledge_critical_alert(
Extension(db): Extension<DB>,
authed: ApiAuthed,
Path(id): Path<i32>,
) -> error::Result<String> {
require_super_admin(&db, &authed.email).await?;
sqlx::query!(
"UPDATE alerts SET acknowledged = true WHERE id = $1",
id
)
.execute(&db)
.await?;
tracing::info!("Acknowledged critical alert with id: {}", id);
Ok("Critical alert acknowledged".to_string())
}
#[cfg(not(feature = "enterprise"))]
pub async fn acknowledge_critical_alert() -> error::Error {
error::Error::NotFound("Critical Alerts require EE".to_string())
}
#[cfg(feature = "enterprise")]
pub async fn acknowledge_all_critical_alerts(
Extension(db): Extension<DB>,
authed: ApiAuthed,
) -> error::Result<String> {
require_super_admin(&db, &authed.email).await?;
sqlx::query!(
"UPDATE alerts SET acknowledged = true WHERE acknowledged = false"
)
.execute(&db)
.await?;
tracing::info!("Acknowledged all unacknowledged critical alerts");
Ok("All unacknowledged critical alerts acknowledged".to_string())
}
#[cfg(not(feature = "enterprise"))]
pub async fn acknowledge_all_critical_alerts() -> error::Error {
error::Error::NotFound("Critical Alerts require EE".to_string())
}
+11 -1
View File
@@ -746,10 +746,20 @@ pub async fn fetch_api_authed(
username_override: String,
) -> error::Result<ApiAuthed> {
let permissioned_as = username_to_permissioned_as(username.as_str());
fetch_api_authed_from_permissioned_as(permissioned_as, email, w_id, db, username_override).await
}
pub async fn fetch_api_authed_from_permissioned_as(
permissioned_as: String,
email: String,
w_id: &str,
db: &DB,
username_override: String,
) -> error::Result<ApiAuthed> {
let authed =
fetch_authed_from_permissioned_as(permissioned_as, email.clone(), w_id, db).await?;
Ok(ApiAuthed {
username: username,
username: authed.username,
email: email,
is_admin: authed.is_admin,
is_operator: authed.is_operator,
+27
View File
@@ -16,6 +16,7 @@ use rand::Rng;
use serde::{Deserialize, Serialize, Serializer};
use crate::{
error::Error,
more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
scripts::{Schema, ScriptHash, ScriptLang},
};
@@ -651,3 +652,29 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
});
}
}
pub async fn has_failure_module<'c>(flow: sqlx::types::Uuid, db: &sqlx::Pool<sqlx::Postgres>, completed: bool) -> Result<bool, Error> {
if completed {
sqlx::query_scalar!(
"SELECT raw_flow->'failure_module' != 'null'::jsonb
FROM completed_job
WHERE id = $1",
flow
)
} else {
sqlx::query_scalar!(
"SELECT raw_flow->'failure_module' != 'null'::jsonb
FROM queue
WHERE id = $1",
flow
)
}
.fetch_one(db)
.await
.map_err(|e| {
Error::InternalErr(format!(
"error during retrieval of has_failure_module: {e:#}"
))
})
.map(|v| v.unwrap_or(false))
}
@@ -29,6 +29,7 @@ pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation
pub const HUB_BASE_URL_SETTING: &str = "hub_base_url";
pub const HUB_ACCESSIBLE_URL_SETTING: &str = "hub_accessible_url";
pub const CRITICAL_ERROR_CHANNELS_SETTING: &str = "critical_error_channels";
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";
+2
View File
@@ -86,6 +86,8 @@ lazy_static::lazy_static! {
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 METRICS_DEBUG_ENABLED: AtomicBool = AtomicBool::new(false);
pub static ref CRITICAL_ALERT_MUTE_UI_ENABLED: AtomicBool = AtomicBool::new(false);
pub static ref BASE_URL: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
+4 -1
View File
@@ -109,10 +109,13 @@ pub struct S3AwsOidcResource {
pub audience: Option<String>,
}
#[derive(Deserialize, Clone)]
#[derive(Serialize, Deserialize, Clone)]
pub struct S3Object {
pub s3: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub storage: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub filename: Option<String>,
}
#[cfg(feature = "parquet")]
+13 -4
View File
@@ -28,6 +28,9 @@ pub const DEFAULT_PER_PAGE: usize = 1000;
pub const GIT_VERSION: &str =
git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
use std::sync::atomic::Ordering;
use crate::CRITICAL_ALERT_MUTE_UI_ENABLED;
lazy_static::lazy_static! {
pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new()
.user_agent("windmill/beta")
@@ -244,9 +247,12 @@ pub fn generate_lock_id(database_name: &str) -> i64 {
pub async fn report_critical_error(error_message: String, _db: DB) -> () {
tracing::error!("CRITICAL ERROR: {error_message}");
let mute = CRITICAL_ALERT_MUTE_UI_ENABLED.load(Ordering::Relaxed);
if let Err(err) = sqlx::query!(
"INSERT INTO alerts (alert_type, message) VALUES ('critical_error', $1)",
error_message
"INSERT INTO alerts (alert_type, message, acknowledged) VALUES ('critical_error', $1, $2)",
error_message,
mute
)
.execute(&_db)
.await
@@ -261,9 +267,12 @@ pub async fn report_critical_error(error_message: String, _db: DB) -> () {
pub async fn report_recovered_critical_error(message: String, _db: DB) -> () {
tracing::info!("RECOVERED CRITICAL ERROR: {message}");
let mute = CRITICAL_ALERT_MUTE_UI_ENABLED.load(Ordering::Relaxed);
if let Err(err) = sqlx::query!(
"INSERT INTO alerts (alert_type, message) VALUES ('recovered_critical_error', $1)",
message
"INSERT INTO alerts (alert_type, message, acknowledged) VALUES ('recovered_critical_error', $1, $2)",
message,
mute
)
.execute(&_db)
.await
+4 -16
View File
@@ -71,6 +71,9 @@ use windmill_common::BASE_URL;
#[cfg(feature = "cloud")]
use windmill_common::users::SUPERADMIN_SYNC_EMAIL;
#[cfg(feature = "enterprise")]
use windmill_common::flows::has_failure_module;
#[cfg(feature = "enterprise")]
use windmill_common::worker::CLOUD_HOSTED;
@@ -482,13 +485,6 @@ where
}
}
#[cfg(feature = "enterprise")]
#[derive(Deserialize)]
struct RawFlowFailureModule {
#[cfg(feature = "enterprise")]
failure_module: Option<Box<RawValue>>,
}
#[instrument(level = "trace", skip_all)]
pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Send>(
db: &Pool<Postgres>,
@@ -952,15 +948,7 @@ pub async fn add_completed_job<
} else if !skip_downstream_error_handlers
&& (matches!(queued_job.job_kind, JobKind::Script)
|| matches!(queued_job.job_kind, JobKind::Flow)
&& queued_job
.raw_flow
.as_ref()
.and_then(|v| {
serde_json::from_str::<RawFlowFailureModule>((**v).get())
.ok()
.and_then(|v| v.failure_module)
})
.is_none())
&& !has_failure_module(job_id, db, true).await.unwrap_or(false))
&& queued_job.parent_job.is_none()
{
let result = serde_json::from_str(
+12 -2
View File
@@ -89,8 +89,18 @@ cat bp | tail -1 >> ./result2.out &
# Run main.sh in the same process group
{bash} ./main.sh "$@" 2>&1 | tee bp &
# Wait for all background processes to finish
wait
pid=$!
# Wait for main.sh to finish and capture its exit status
wait $pid
exit_status=$?
# Clean up the named pipe and background processes
rm -f bp
# Exit with the captured status
exit $exit_status
"#,
bash = BIN_BASH.as_str(),
);
+1
View File
@@ -799,6 +799,7 @@ pub async fn get_cached_resource_value_if_valid(
S3Object {
s3: s3_file_key.clone(),
storage: cached_resource.storage.clone(),
filename: None,
},
)
.await;
+10 -2
View File
@@ -48,6 +48,7 @@ use futures::{
use crate::common::{resolve_job_timeout, OccupancyMetrics};
use crate::job_logger::{append_job_logs, append_with_limit, LARGE_LOG_THRESHOLD_SIZE};
use crate::job_logger_ee::process_streaming_log_lines;
use crate::{MAX_RESULT_SIZE, MAX_WAIT_FOR_SIGINT, MAX_WAIT_FOR_SIGTERM};
lazy_static::lazy_static! {
@@ -434,6 +435,8 @@ pub async fn handle_child(
}
}
async fn get_mem_peak(pid: Option<u32>, nsjail: bool) -> i32 {
if pid.is_none() {
return -1;
@@ -689,19 +692,24 @@ fn child_joined_output_stream(
let stdout = BufReader::new(stdout).lines();
let stderr = BufReader::new(stderr).lines();
stream::select(lines_to_stream(stderr), lines_to_stream(stdout))
stream::select(lines_to_stream(stderr, true), lines_to_stream(stdout, false))
}
pub fn lines_to_stream<R: tokio::io::AsyncBufRead + Unpin>(
mut lines: tokio::io::Lines<R>,
stderr: bool,
) -> impl futures::Stream<Item = io::Result<String>> {
stream::poll_fn(move |cx| {
std::pin::Pin::new(&mut lines)
.poll_next_line(cx)
.map(|result| result.transpose())
.map(|result| {
process_streaming_log_lines(result, stderr)
})
})
}
pub fn process_status(status: ExitStatus) -> error::Result<()> {
if status.success() {
Ok(())
+9 -190
View File
@@ -1,15 +1,6 @@
use itertools::Itertools;
use swc_ecma_parser::lexer::util::CharExt;
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use object_store::path::Path;
use regex::Regex;
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS;
use windmill_common::error::{self};
use windmill_common::worker::{CLOUD_HOSTED, TMP_DIR};
use windmill_common::worker::CLOUD_HOSTED;
use windmill_queue::append_logs;
@@ -19,6 +10,12 @@ use std::sync::Arc;
use uuid::Uuid;
use windmill_common::DB;
use crate::job_logger_ee::default_disk_log_storage;
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use crate::job_logger_ee::s3_storage;
pub enum CompactLogs {
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
NotEE,
@@ -28,150 +25,7 @@ pub enum CompactLogs {
S3,
}
async fn compact_logs(
job_id: Uuid,
w_id: &str,
db: &DB,
nlogs: String,
total_size: Arc<AtomicU32>,
compact_kind: CompactLogs,
_worker_name: &str,
) -> error::Result<(String, String)> {
let mut prev_logs = sqlx::query_scalar!(
"SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2",
job_id,
w_id
)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or_default();
let size = prev_logs.char_indices().count() as i32;
let nlogs_len = nlogs.char_indices().count();
let to_keep_in_db = usize::max(
usize::min(nlogs_len, 3000),
nlogs_len % LARGE_LOG_THRESHOLD_SIZE,
);
let extra_split = to_keep_in_db < nlogs_len;
let stored_in_storage_len = if extra_split {
nlogs_len - to_keep_in_db
} else {
0
};
let extra_to_newline = nlogs
.chars()
.skip(stored_in_storage_len)
.find_position(|x| x.is_line_break())
.map(|(i, _)| i)
.unwrap_or(to_keep_in_db);
let stored_in_storage_to_newline = stored_in_storage_len + extra_to_newline;
let (append_to_storage, stored_in_db) = if extra_split {
if stored_in_storage_to_newline == nlogs.len() {
(nlogs.as_ref(), "".to_string())
} else {
let split_idx = nlogs
.char_indices()
.nth(stored_in_storage_to_newline)
.map(|(i, _)| i)
.unwrap_or(0);
let (append_to_storage, stored_in_db) = nlogs.split_at(split_idx);
// tracing::error!("{append_to_storage} ||||| {stored_in_db}");
// tracing::error!(
// "{:?} {:?} {} {}",
// excess_prev_logs.lines().last(),
// current_logs.lines().next(),
// split_idx,
// excess_size_modulo
// );
(append_to_storage, stored_in_db.to_string())
}
} else {
// tracing::error!("{:?}", nlogs.lines().last());
("", nlogs.to_string())
};
let new_size_with_excess = size + stored_in_storage_to_newline as i32;
let new_size = total_size.fetch_add(
new_size_with_excess as u32,
std::sync::atomic::Ordering::SeqCst,
) + new_size_with_excess as u32;
let path = format!(
"logs/{job_id}/{}_{new_size}.txt",
chrono::Utc::now().timestamp_millis()
);
let mut new_current_logs = match compact_kind {
CompactLogs::NoS3 => format!("\n[windmill] No object storage set in instance settings. Previous logs have been saved to disk at {path}"),
CompactLogs::S3 => format!("\n[windmill] Previous logs have been saved to object storage at {path}"),
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
CompactLogs::NotEE => format!("\n[windmill] Previous logs have been saved to disk at {path}"),
};
new_current_logs.push_str(&stored_in_db);
sqlx::query!(
"UPDATE job_logs SET logs = $1, log_offset = $2,
log_file_index = array_append(coalesce(log_file_index, array[]::text[]), $3)
WHERE workspace_id = $4 AND job_id = $5",
new_current_logs,
new_size as i32,
path,
w_id,
job_id
)
.execute(db)
.await?;
prev_logs.push_str(&append_to_storage);
return Ok((prev_logs, path));
}
async fn default_disk_log_storage(
job_id: Uuid,
w_id: &str,
db: &DB,
nlogs: String,
total_size: Arc<AtomicU32>,
compact_kind: CompactLogs,
worker_name: &str,
) {
match compact_logs(
job_id,
&w_id,
&db,
nlogs,
total_size,
compact_kind,
worker_name,
)
.await
{
Err(e) => tracing::error!("Could not compact logs for job {job_id}: {e:?}",),
Ok((prev_logs, path)) => {
let path = format!("{}/{}", TMP_DIR, path);
let splitted = &path.split("/").collect_vec();
tokio::fs::create_dir_all(splitted.into_iter().take(splitted.len() - 1).join("/"))
.await
.map_err(|e| {
tracing::error!("Could not create logs directory: {e:?}",);
e
})
.ok();
let created = tokio::fs::File::create(&path).await;
if let Err(e) = created {
tracing::error!("Could not create logs file {path}: {e:?}",);
return;
}
if let Err(e) = tokio::fs::write(&path, prev_logs).await {
tracing::error!("Could not write to logs file {path}: {e:?}");
} else {
tracing::info!("Logs length of {job_id} has exceeded a threshold. Previous logs have been saved to disk at {path}");
}
}
}
}
pub(crate) async fn append_job_logs(
job_id: Uuid,
@@ -184,43 +38,7 @@ pub(crate) async fn append_job_logs(
) -> () {
if must_compact_logs {
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if let Some(os) = OBJECT_STORE_CACHE_SETTINGS.read().await.clone() {
match compact_logs(
job_id,
&w_id,
&db,
logs,
total_size,
CompactLogs::S3,
&worker_name,
)
.await
{
Err(e) => tracing::error!("Could not compact logs for job {job_id}: {e:?}",),
Ok((prev_logs, path)) => {
tracing::info!("Logs length of {job_id} has exceeded a threshold. Previous logs have been saved to object storage at {path}");
let path2 = path.clone();
if let Err(e) = os
.put(&Path::from(path), prev_logs.to_string().into_bytes().into())
.await
{
tracing::error!("Could not save logs to s3: {e:?}");
}
tracing::info!("Logs of {job_id} saved to object storage at {path2}");
}
}
} else {
default_disk_log_storage(
job_id,
&w_id,
&db,
logs,
total_size,
CompactLogs::NoS3,
&worker_name,
)
.await;
}
s3_storage(job_id, &w_id, &db, logs, total_size, &worker_name).await;
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
{
@@ -252,6 +70,7 @@ pub fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize) {
if *NO_LOGS_AT_ALL {
return;
}
let src_str;
let src = {
src_str = RE_00.replace_all(src, "");
@@ -0,0 +1,30 @@
use std::io;
use std::sync::atomic::AtomicU32;
use std::sync::Arc;
use uuid::Uuid;
use windmill_common::DB;
use crate::job_logger::CompactLogs;
#[cfg(all(feature = "enterprise", feature = "parquet"))]
pub(crate) async fn s3_storage(_job_id: Uuid, _w_id: &String, _db: &sqlx::Pool<sqlx::Postgres>, _logs: &String, _total_size: &Arc<AtomicU32>, _worker_name: &String) {
tracing::info!("Logs length of {job_id} has exceeded a threshold. Implementation to store excess on s3 in not OSS");
}
pub(crate) async fn default_disk_log_storage(
job_id: Uuid,
_w_id: &str,
_db: &DB,
_nlogs: String,
_total_size: Arc<AtomicU32>,
_compact_kind: CompactLogs,
_worker_name: &str,
) {
tracing::info!("Logs length of {job_id} has exceeded a threshold. Implementation to store excess on disk in not OSS");
}
pub(crate) fn process_streaming_log_lines(r: Result<Option<String>, io::Error>, _stderr: bool) -> Option<Result<String, io::Error>> {
r.transpose()
}
+1 -1
View File
@@ -29,7 +29,7 @@ mod rust_executor;
mod worker;
mod worker_flow;
mod worker_lockfiles;
mod job_logger_ee;
pub use worker::*;
pub use result_processor::handle_job_error;
+11 -4
View File
@@ -199,8 +199,11 @@ pub async fn uv_pip_compile(
.clone()
.map(handle_ephemeral_token);
if let Some(url) = pip_extra_index_url.as_ref() {
args.extend(["--extra-index-url", url, "--no-emit-index-url"]);
pip_args.push(format!("--extra-index-url {}", url));
url.split(",").for_each(|url| {
args.extend(["--extra-index-url", url]);
pip_args.push(format!("--extra-index-url {}", url));
});
args.push("--no-emit-index-url");
}
let pip_index_url = PIP_INDEX_URL
.read()
@@ -279,7 +282,9 @@ pub async fn uv_pip_compile(
.clone()
.map(handle_ephemeral_token);
if let Some(url) = pip_extra_index_url.as_ref() {
args.extend(["--extra-index-url", url]);
url.split(",").for_each(|url| {
args.extend(["--extra-index-url", url]);
});
}
let pip_index_url = PIP_INDEX_URL
.read()
@@ -1167,7 +1172,9 @@ pub async fn handle_python_reqs(
.map(handle_ephemeral_token);
if let Some(url) = pip_extra_index_url.as_ref() {
command_args.extend(["--extra-index-url", url]);
url.split(",").for_each(|url| {
command_args.extend(["--extra-index-url", url]);
});
}
let pip_index_url = PIP_INDEX_URL
.read()
@@ -295,6 +295,7 @@ pub async fn process_result(
message: format!("error during execution of the script:\n{}", err),
name: "ExecutionErr".to_string(),
step_id: job.flow_step_id.clone(),
exit_code: None,
}),
};
@@ -608,6 +609,8 @@ pub struct SerializedError {
pub name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub step_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub exit_code: Option<i32>,
}
pub fn extract_error_value(log_lines: &str, i: i32, step_id: Option<String>) -> Box<RawValue> {
return to_raw_value(&SerializedError {
@@ -617,5 +620,6 @@ pub fn extract_error_value(log_lines: &str, i: i32, step_id: Option<String>) ->
),
name: "ExecutionErr".to_string(),
step_id,
exit_code: Some(i),
});
}
+2 -19
View File
@@ -49,7 +49,7 @@ use windmill_common::{
Approval, BranchAllStatus, BranchChosen, FlowStatus, FlowStatusModule, RetryStatus,
MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
flows::{has_failure_module, FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
};
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
@@ -963,7 +963,7 @@ pub async fn update_flow_status_after_job_completion_internal<
false
if !is_failure_step
&& !skip_error_handler
&& has_failure_module(flow, db).await? =>
&& has_failure_module(flow, db, false).await? =>
{
true
}
@@ -1290,23 +1290,6 @@ async fn compute_skip_branchall_failure<'c>(
}))
}
async fn has_failure_module<'c>(flow: Uuid, db: &DB) -> Result<bool, Error> {
sqlx::query_scalar::<_, Option<bool>>(
"SELECT raw_flow->'failure_module' != 'null'::jsonb
FROM queue
WHERE id = $1",
)
.bind(flow)
.fetch_one(db)
.await
.map_err(|e| {
Error::InternalErr(format!(
"error during retrieval of has_failure_module: {e:#}"
))
})
.map(|v| v.unwrap_or(false))
}
// async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result<FlowCleanupModule, Error> {
// tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid);
// let raw_value = sqlx::query_scalar!(
+13 -3
View File
@@ -8,6 +8,7 @@ import { DenoLandProvider } from "https://deno.land/x/cliffy@v0.25.7/command/upg
import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts";
import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts";
import { VERSION, createBenchScript, getFlowPayload, login } from "./lib.ts";
@@ -172,8 +173,8 @@ export async function main({
kind: "script",
path: "f/benchmarks/" + kind,
});
} else if (["2steps"].includes(kind)) {
nStepsFlow = 2;
} else if (["2steps", "bigscriptinflow"].includes(kind)) {
nStepsFlow = kind == "2steps" ? 2 : 1;
const payload = getFlowPayload(kind);
body = JSON.stringify({
kind: "flow",
@@ -194,6 +195,15 @@ export async function main({
kind: "script",
path: kind.substr(7),
});
} else if (kind == "bigrawscript") {
noVerify = true;
body = JSON.stringify({
kind: "rawscript",
rawscript: {
language: api.RawScript.language.BASH,
content: "# let's bloat that bash script, 3.. 2.. 1.. BOOM\n".repeat(25000) + "echo \"$WM_FLOW_JOB_ID\"\n",
},
});
} else {
throw new Error("Unknown script pattern " + kind);
}
@@ -246,7 +256,7 @@ export async function main({
} else {
const elapsed = start ? Date.now() - start : 0;
completedJobs = await getCompletedJobsCount();
if (kind === "2steps" || kind.startsWith("flow:")) {
if (nStepsFlow > 0) {
completedJobs = Math.floor(completedJobs / (nStepsFlow + 1));
}
const avgThr = ((completedJobs / elapsed) * 1000).toFixed(2);
+18 -1
View File
@@ -2,7 +2,7 @@ import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts";
import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts";
export const VERSION = "v1.419.0";
export const VERSION = "v1.423.2";
export async function login(email: string, password: string): Promise<string> {
return await windmill.UserService.login({
@@ -235,6 +235,23 @@ export const getFlowPayload = (flowPattern: string): api.FlowPreview => {
],
},
};
} else if (flowPattern == "bigscriptinflow") {
return {
path: "bigscriptinflow",
args: {},
value: {
modules: [
{
value: {
input_transforms: {},
language: api.RawScript.language.BASH,
type: "rawscript",
content: "# let's bloat that bash script, 3.. 2.. 1.. BOOM\n".repeat(25000) + "echo \"$WM_FLOW_JOB_ID\"\n",
},
}
],
}
};
} else {
return {
path: "2steps",
+8
View File
@@ -30,5 +30,13 @@
{
"kind": "nativets",
"jobs": 2000
},
{
"kind": "bigrawscript",
"jobs": 10000
},
{
"kind": "bigscriptinflow",
"jobs": 10000
}
]
+1 -1
View File
@@ -60,7 +60,7 @@ export {
// }
// });
export const VERSION = "1.419.0";
export const VERSION = "1.423.2";
const command = new Command()
.name("wmill")
+56 -39
View File
@@ -13,31 +13,23 @@
overlays = [ (import rust-overlay) ];
};
buildInputs = with pkgs; [
openssl.dev libxml2.dev xmlsec.dev libxslt.dev
rust-bin.beta.latest.default nodejs go
xcaddy sqlx-cli pkg-config
openssl openssl.dev libxml2.dev xmlsec.dev libxslt.dev
rust-bin.stable.latest.default nodejs
postgresql
pkg-config
];
environment = {
NODE_ENV = "development";
NODE_OPTIONS = "--max-old-space-size=16384";
DATABASE_URL = "postgres://postgres:changeme@127.0.0.1:5432/";
PKG_CONFIG_PATH = pkgs.lib.makeSearchPath "lib/pkgconfig" (with pkgs; [
openssl.dev
libxml2.dev
xmlsec.dev
libxslt.dev
]);
};
rustcTarget = pkgs.hostPlatform.rust.rustcTarget;
fetch_librusty_v8 = { version, shas }: pkgs.fetchurl {
name = "librusty_v8-${version}";
url = "https://github.com/denoland/rusty_v8/releases/download/v${version}/librusty_v8_release_${rustcTarget}.a.gz";
sha256 = shas.${system};
};
PKG_CONFIG_PATH = pkgs.lib.makeSearchPath "lib/pkgconfig" (with pkgs; [
openssl.dev
libxml2.dev
xmlsec.dev
libxslt.dev
]);
in {
devShell = pkgs.mkShell {
inherit (environment) NODE_ENV NODE_OPTIONS DATABASE_URL PKG_CONFIG_PATH;
buildInputs = buildInputs;
buildInputs = buildInputs ++ (with pkgs; [
git xcaddy sqlx-cli flock rust-analyzer
deno python3 python3Packages.pip go bun uv
]);
packages = [
(pkgs.writeScriptBin "wm-caddy" ''
cd ./frontend
@@ -54,34 +46,55 @@
'')
(pkgs.writeScriptBin "wm-migrate" ''
cd ./backend
sqlx database reset
sqlx database create
sqlx migrate run
'')
(pkgs.writeScriptBin "wm-setup" ''
sqlx database create
wm-build
wm-caddy
wm-migrate
'')
(pkgs.writeScriptBin "wm-reset" ''
sqlx database drop -f
sqlx database create
wm-migrate
'')
(pkgs.writeScriptBin "wm-bench" ''
deno run -A benchmarks/main.ts -e admin@windmill.dev -p changeme $*
'')
(pkgs.writeScriptBin "wm" ''
cd ./frontend
npm run dev $*
'')
];
inherit PKG_CONFIG_PATH;
NODE_ENV = "development";
NODE_OPTIONS = "--max-old-space-size=16384";
DATABASE_URL = "postgres://postgres:changeme@127.0.0.1:5432/";
REMOTE = "http://127.0.0.1:8000";
REMOTE_LSP = "http://127.0.0.1:3001";
DENO_PATH = "${pkgs.deno}/bin/deno";
PYTHON_PATH = "${pkgs.python3}/bin/python3";
GO_PATH = "${pkgs.go}/bin/go";
BUN_PATH = "${pkgs.bun}/bin/bun";
UV_PATH = "${pkgs.uv}/bin/uv";
FLOCK_PATH = "${pkgs.flock}/bin/flock";
};
packages.default = self.packages.${system}.windmill;
packages.windmill-client = pkgs.stdenv.mkDerivation {
pname = "windmill-client";
version = (pkgs.lib.strings.trim (builtins.readFile ./version.txt));
src = ./.;
buildInputs = with pkgs; [ nodejs go ];
buildInputs = with pkgs; [ nodejs ];
buildPhase = ''
export HOME=$(pwd)
npm config set strict-ssl false
cd frontend
npm install --verbose
npm run generate-backend-client-mac
npm run ${if pkgs.stdenv.isDarwin then "generate-backend-client-mac" else "generate-backend-client"}
NODE_OPTIONS="--max-old-space-size=8192" npm run build
'';
@@ -91,8 +104,6 @@
'';
};
packages.windmill = pkgs.rustPlatform.buildRustPackage {
inherit (environment) DATABASE_URL PKG_CONFIG_PATH;
pname = "windmill";
version = (pkgs.lib.strings.trim (builtins.readFile ./version.txt));
@@ -111,26 +122,32 @@
};
};
buildFeatures = [ "enterprise" ];
buildFeatures = [ ];
doCheck = false;
preBuild = ''
export HOME=$(pwd)
npm config set strict-ssl false
cd backend
sqlx database create
sqlx migrate run
'';
inherit PKG_CONFIG_PATH;
SQLX_OFFLINE = true;
FRONTEND_BUILD_DIR = "${self.packages.${system}.windmill-client}/build";
RUSTY_V8_ARCHIVE = (fetch_librusty_v8 {
version = "130.0.1";
shas = {
x86_64-linux = pkgs.lib.fakeHash;
aarch64-linux = pkgs.lib.fakeHash;
x86_64-darwin = pkgs.lib.fakeHash;
aarch64-darwin = "sha256-d1QTLt8gOUFxACes4oyIYgDF/srLOEk+5p5Oj1ECajQ=";
RUSTY_V8_ARCHIVE =
let
version = "130.0.1";
target = pkgs.hostPlatform.rust.rustcTarget;
sha256 = {
x86_64-linux = pkgs.lib.fakeHash;
aarch64-linux = pkgs.lib.fakeHash;
x86_64-darwin = pkgs.lib.fakeHash;
aarch64-darwin = "sha256-d1QTLt8gOUFxACes4oyIYgDF/srLOEk+5p5Oj1ECajQ=";
}.${system};
in pkgs.fetchurl {
name = "librusty_v8-${version}";
url = "https://github.com/denoland/rusty_v8/releases/download/v${version}/librusty_v8_release_${target}.a.gz";
inherit sha256;
};
});
};
});
}
+44
View File
@@ -1,5 +1,7 @@
# Developing
[Using Nix](#nix), otherwise:
In the `frontend/` directory:
- install the dependencies with `npm install` (or `pnpm install` or `yarn`)
@@ -136,3 +138,45 @@ NOTCATCHALL=true npm run build
which will generate an index.html and allow you to serve the frontend with any static server.
Env variables used for build are set in .env file. See [https://vitejs.dev/guide/env-and-mode.html#env-files](https://vitejs.dev/guide/env-and-mode.html#env-files) for more details.
## Nix
Windmill use [nix flakes](https://nixos.wiki/wiki/Flakes):
```bash
nix run github:windmill-labs/windmill
```
### Development
```bash
nix develop # enter a dev shell containing all necessary packages.
wm-setup # build the frontend and setup the database.
wm # run the frontend.
# In an other shell:
nix develop
cd backend
cargo run
```
### Updating [`flake.nix`](../flake.nix)
```bash
nix flake update # update the lock file.
```
Some cargo dependencies use fixed git revisions, which are also fixed in `flake.nix`:
```nix
outputHashes = {
"php-parser-rs-0.1.3" = "sha256-ZeI3KgUPmtjlRfq6eAYveqt8Ay35gwj6B9iOQRjQa9A=";
# ...
};
```
When updating a revision, replace the incorrect `sha256` with `pkgs.lib.fakeHash`:
```diff
- "php-parser-rs-0.1.3" = "sha256-ZeI3KgUPmtjlRfq6eAYveqt8Ay35gwj6B9iOQRjQa9A=";
+ "php-parser-rs-0.1.3" = pkgs.lib.fakeHash;
```
Then run `nix build .#windmill` and update with the correct sha.
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "windmill-components",
"version": "1.419.0",
"version": "1.423.2",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "windmill-components",
"version": "1.419.0",
"version": "1.423.2",
"license": "AGPL-3.0",
"dependencies": {
"@aws-crypto/sha256-js": "^4.0.0",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "windmill-components",
"version": "1.419.0",
"version": "1.423.2",
"scripts": {
"dev": "vite dev",
"build": "vite build",
+1 -1
View File
@@ -98,7 +98,7 @@
}}><Expand size={18} /></button
>
{/if}
<div class="max-h-40 overflow-auto">
<div class="max-h-60 overflow-auto">
<ObjectViewer collapsed={false} topBrackets={true} pureViewer={true} json={value} />
</div>
</div>
+22 -8
View File
@@ -418,6 +418,9 @@
bind:selected={value}
options={itemsType?.multiselect ?? []}
selectedOptionsDraggable={true}
on:open={() => {
dispatch('focus')
}}
/>
</div>
{:else if itemsType?.enum != undefined && Array.isArray(itemsType?.enum) && Array.isArray(value)}
@@ -428,6 +431,9 @@
bind:selected={value}
options={itemsType?.enum ?? []}
selectedOptionsDraggable={true}
on:open={() => {
dispatch('focus')
}}
/>
</div>
{:else}
@@ -816,6 +822,12 @@
bind:value
{enum_}
{autofocus}
on:focus={() => {
dispatch('focus')
}}
on:blur={(e) => {
dispatch('blur')
}}
enumLabels={extra['enumLabels']}
/>
</div>
@@ -884,14 +896,16 @@
<div class="flex flex-col w-full">
<div class="flex flex-row w-full items-center justify-between relative">
{#if password || extra?.['password'] == true}
{#if value && typeof value == 'string' && value?.startsWith('$var:')}
<input type="text" bind:value />
{:else if onlyMaskPassword}
<Password
{disabled}
bind:password={value}
placeholder={placeholder ?? defaultValue ?? ''}
/>
{#if onlyMaskPassword}
{#if value && typeof value == 'string' && value?.startsWith('$var:')}
<input type="text" bind:value />
{:else}
<Password
{disabled}
bind:password={value}
placeholder={placeholder ?? defaultValue ?? ''}
/>
{/if}
{:else}
<PasswordArgInput {disabled} bind:value />
{/if}
@@ -349,6 +349,7 @@
{isAppInput}
on:change={() => {
schema = schema
// console.log('schema', schema)
dispatch('change', schema)
}}
>
@@ -7,6 +7,7 @@
import FlowGraphViewerStep from './FlowGraphViewerStep.svelte'
import FlowGraphV2 from './graph/FlowGraphV2.svelte'
import { dfs } from './flows/dfs'
import { workspaceStore } from '$lib/stores'
export let flow: {
summary: string
@@ -22,6 +23,7 @@
export let noGraph = false
export let triggerNode = false
export let stepDetail: FlowModule | string | undefined = undefined
export let workspace: string | undefined = $workspaceStore
const dispatch = createEventDispatcher()
</script>
@@ -37,6 +39,7 @@
path={flow?.path}
{download}
minHeight={400}
{workspace}
modules={flow?.value?.modules}
failureModule={flow?.value?.failure_module}
preprocessorModule={flow?.value?.preprocessor_module}
@@ -29,6 +29,7 @@
import Alert from './common/alert/Alert.svelte'
import FlowGraphViewerStep from './FlowGraphViewerStep.svelte'
import FlowGraphV2 from './graph/FlowGraphV2.svelte'
import { buildPrefix } from './graph/graphBuilder'
const dispatch = createEventDispatcher()
@@ -74,6 +75,10 @@
export let wideResults = false
export let hideFlowResult = false
export let workspace: string | undefined = $workspaceStore
export let prefix: string | undefined = undefined
export let subflowParentsGlobalModuleStates: Writable<Record<string, GraphModuleState>>[] = []
export let subflowParentsDurationStatuses: Writable<Record<string, DurationStatus>>[] = []
let jobResults: any[] =
flowJobIds?.flowJobs?.map((x, id) => `iter #${id + 1} not loaded by frontend yet`) ?? []
@@ -83,6 +88,7 @@
let localModuleStates: Writable<Record<string, GraphModuleState>> = writable({})
let localDurationStatuses: Writable<Record<string, DurationStatus>> = writable({})
let expandedSubflows: Record<string, FlowModule[]> = {}
export let job: Job | undefined = undefined
@@ -94,6 +100,24 @@
// }
// }
function updateModuleStates(
moduleState: Writable<Record<string, GraphModuleState>>,
key: string,
newValue: GraphModuleState,
keepType: boolean | undefined
) {
moduleState.update((x) => {
if (keepType && (x[key]?.type == 'Success' || x[key]?.type == 'Failure')) {
newValue.type = x[key].type
}
x[key] = newValue
return x
})
}
function buildSubflowKey(key: string, prefix: string | undefined) {
return prefix ? 'subflow:' + prefix + key : key
}
function setModuleState(
key: string,
value: Partial<GraphModuleState>,
@@ -102,15 +126,14 @@
) {
let newValue = { ...($localModuleStates[key] ?? {}), ...value }
if (!deepEqual($localModuleStates[key], value) || force) {
;[localModuleStates, ...globalModuleStates].forEach((s) => {
s.update((x) => {
if (keepType && (x[key]?.type == 'Success' || x[key]?.type == 'Failure')) {
newValue.type = x[key].type
}
x[key] = newValue
return x
})
})
;[localModuleStates, ...globalModuleStates].forEach((s) =>
updateModuleStates(s, key, newValue, keepType)
)
if (prefix) {
subflowParentsGlobalModuleStates.forEach((s) =>
updateModuleStates(s, buildSubflowKey(key, prefix), newValue, keepType)
)
}
}
}
@@ -124,6 +147,14 @@
return x
})
})
if (prefix) {
subflowParentsDurationStatuses.forEach((s) => {
s.update((x) => {
x[buildSubflowKey(key, prefix)].byJob[id] = value
return x
})
})
}
}
}
@@ -131,15 +162,25 @@
if ($localDurationStatuses[modId] == undefined) {
$localDurationStatuses[modId] = { byJob: {} }
}
let prefixed = modId
globalDurationStatuses.forEach((x) =>
x.update((x) => {
if (x[prefixed] == undefined) {
x[prefixed] = { byJob: {} }
if (x[modId] == undefined) {
x[modId] = { byJob: {} }
}
return x
})
)
if (prefix) {
subflowParentsDurationStatuses.forEach((x) =>
x.update((x) => {
let key = buildSubflowKey(modId, prefix)
if (x[key] == undefined) {
x[key] = { byJob: {} }
}
return x
})
)
}
}
let innerModules: FlowStatusModule[] = []
@@ -368,7 +409,7 @@
if (globalRefreshes) {
let modId = flowJobIds?.moduleId
if (modId) {
globalRefreshes[modId] = async (loopJob) => {
globalRefreshes[buildSubflowKey(modId, prefix)] = async (loopJob) => {
setIteration(loopJob.index, loopJob.job, false, modId ?? '')
refresh(true, loopJob)
}
@@ -631,6 +672,36 @@
})
}
}
function allModulesForTimeline(
modules: FlowModule[],
expandedSubflows: Record<string, FlowModule[]>
): string[] {
const ids = dfs(modules, (x) => x.id)
function rec(ids: string[], prefix: string | undefined): string[] {
return ids.concat(
ids.flatMap((id) => {
let fms = expandedSubflows[id]
let oid = id.split(':').pop()
if (!oid) {
return []
}
let nprefix = buildPrefix(prefix, oid)
return fms
? rec(
dfs(fms, (x) =>
x.id.startsWith('subflow:') ? x.id : buildSubflowKey(x.id, nprefix)
),
nprefix
)
: []
})
)
}
return rec(ids, undefined)
}
</script>
{#if notAnonynmous}
@@ -803,6 +874,9 @@
job={storedListJobs[j]}
globalModuleStates={[localModuleStates, ...globalModuleStates]}
globalDurationStatuses={[localDurationStatuses, ...globalDurationStatuses]}
{prefix}
{subflowParentsGlobalModuleStates}
{subflowParentsDurationStatuses}
render={forloop_selected == loopJobId && selected == 'sequence' && render}
reducedPolling={flowJobIds?.flowJobs.length && flowJobIds?.flowJobs.length > 20}
{workspaceId}
@@ -880,6 +954,9 @@
{childFlow}
globalModuleStates={[localModuleStates, ...globalModuleStates]}
globalDurationStatuses={[localDurationStatuses, ...globalDurationStatuses]}
{prefix}
{subflowParentsGlobalModuleStates}
{subflowParentsDurationStatuses}
render={failedRetry == retry_selected && render}
reducedPolling={false}
{workspaceId}
@@ -895,6 +972,17 @@
bind:refresh={recursiveRefresh[mod.job ?? '']}
globalModuleStates={[]}
globalDurationStatuses={[]}
prefix={buildPrefix(prefix, mod.id ?? '')}
subflowParentsGlobalModuleStates={[
localModuleStates,
...globalModuleStates,
...subflowParentsGlobalModuleStates
]}
subflowParentsDurationStatuses={[
localDurationStatuses,
...globalDurationStatuses,
...subflowParentsDurationStatuses
]}
render={selected == 'sequence' && render}
{workspaceId}
jobId={mod.job}
@@ -915,6 +1003,9 @@
globalDurationStatuses={[localDurationStatuses, ...globalDurationStatuses]}
render={selected == 'sequence' && render}
{workspaceId}
{prefix}
{subflowParentsGlobalModuleStates}
{subflowParentsDurationStatuses}
jobId={mod.job}
innerModule={mod.flow_jobs ? job.raw_flow?.modules[i]?.value : undefined}
flowJobIds={mod.flow_jobs
@@ -973,6 +1064,7 @@
minHeight={wrapperHeight}
success={jobId != undefined && isSuccess(job?.['success'])}
flowModuleStates={$localModuleStates}
bind:expandedSubflows
on:select={(e) => {
if (rightColumnSelect != 'node_definition') {
rightColumnSelect = 'node_status'
@@ -1004,11 +1096,13 @@
selectedForloopIndex: detail.index
})
globalRefreshes[detail.moduleId]?.({ job: detail.id, index: detail.index })
// console.log('selectedIteration', prefix, detail.moduleId, globalRefreshes)
}}
modules={job.raw_flow?.modules ?? []}
failureModule={job.raw_flow?.failure_module}
preprocessorModule={job.raw_flow?.preprocessor_module}
allowSimplifiedPoll={false}
{workspace}
/>
</div>
<div
@@ -1033,7 +1127,10 @@
aggregateWaitTime={job?.aggregate_wait_time_ms}
flowDone={job?.['success'] != undefined}
bind:this={flowTimeline}
flowModules={dfs(job.raw_flow?.modules ?? [], (x) => x.id)}
flowModules={allModulesForTimeline(
job?.raw_flow?.modules ?? [],
expandedSubflows ?? {}
)}
durationStatuses={localDurationStatuses}
/>
{:else if rightColumnSelect == 'node_status'}
@@ -178,7 +178,7 @@
{/if}
<div class="px-2 py-2 grid grid-cols-6 w-full">
<div class="truncate">{k}</div>
<div class="truncate">{k.startsWith('subflow:') ? k.substring(8) : k}</div>
<div class="col-span-5 flex min-h-6">
{#if min && total}
<div class="flex flex-col gap-2 w-full p-2 ml-4">
@@ -238,7 +238,7 @@
setPropertyType(arg?.value)
codeInjectionDetected = checkCodeInjection(arg?.value) != undefined
} else if (propertyType == 'javascript' && focused) {
setPropertyType(arg?.expr)
// setPropertyType(arg?.expr)
$inputMatches = checkCodeInjection(arg?.expr)
}
}
@@ -279,7 +279,7 @@
class={twMerge(
'pl-2 pt-2 pb-2 ml-2 relative hover:bg-surface hover:shadow-md transition-all duration-200',
$propPickerConfig?.propName == argName
? 'bg-surface border-l-4 border-blue-500 shadow-md rounded-l-md z-2000'
? 'bg-surface border-l-4 border-blue-500 shadow-md rounded-l-md z-50 '
: 'hover:rounded-md',
$$props.class
)}
@@ -565,10 +565,13 @@
</div>
{#if $propPickerConfig?.propName == argName}
<div class="text-blue-500 mt-2" in:fade={{ duration: 200 }}>
<div
class="text-blue-500 absolute top-2 lg:-right-2.5 -right-1"
in:fade={{ duration: 200 }}
>
<svg
xmlns="http://www.w3.org/2000/svg"
width="14"
width="16"
height="24"
viewBox="0 0 24 24"
fill="currentColor"
@@ -86,7 +86,7 @@
{#if keys.length > 0}
{#each keys as argName (argName)}
{#if (!filter || filter.includes(argName)) && Object.keys(schema.properties ?? {}).includes(argName)}
<div class="z-10 pt-2 relative">
<div class="pt-2 relative">
<InputTransformForm
{previousModuleId}
bind:arg={args[argName]}
@@ -61,6 +61,18 @@
export let render = true
export let title: string | undefined = undefined
export let placeholder: string | undefined = undefined
export let appPath: string | undefined = undefined
export let computeS3ForceViewerPolicies:
| (() =>
| {
allowed_resources: string[]
allow_user_resources: boolean
allow_workspace_resource: boolean
file_key_regex: string
}
| undefined)
| undefined = undefined
export let workspace: string | undefined = undefined
let oneOfSelected: string | undefined = undefined
async function updateOneOfSelected(oneOf: SchemaProperty[] | undefined) {
@@ -428,6 +440,9 @@
.toLowerCase() == 's3object'}
<div class="flex flex-col w-full gap-1">
<FileUpload
{appPath}
computeForceViewerPolicies={computeS3ForceViewerPolicies}
{workspace}
allowMultiple={false}
randomFileKey={true}
on:addition={(evt) => {
@@ -16,6 +16,18 @@
export let defaultValues: Record<string, any> = {}
export let dynamicEnums: Record<string, any> = {}
export let disabled: boolean = false
export let appPath: string | undefined = undefined
export let computeS3ForceViewerPolicies:
| (() =>
| {
allowed_resources: string[]
allow_user_resources: boolean
allow_workspace_resource: boolean
file_key_regex: string
}
| undefined)
| undefined = undefined
export let workspace: string | undefined = undefined
let inputCheck: { [id: string]: boolean } = {}
let errors: { [id: string]: string } = {}
@@ -104,6 +116,9 @@
{displayType}
{css}
disabled={disabled || schema.properties[argName].disabled}
{appPath}
{computeS3ForceViewerPolicies}
{workspace}
/>
{/if}
{/each}
@@ -10,41 +10,53 @@
let path = ''
let password = value && !value.startsWith('$var:') ? value : ''
let isGenerating = false
async function generateValue() {
let npath =
'u/' +
($userStore?.username ?? $userStore?.email)?.split('@')[0] +
'/secret_arg/' +
generateRandomString(12)
let nvalue = '$var:' + npath
await VariableService.createVariable({
workspace: $workspaceStore!,
requestBody: {
value: password,
is_secret: true,
path: npath,
description: '',
expires_at: new Date(Date.now() + 1000 * 60 * 60 * 24 * 7).toISOString()
}
})
path = npath
value = nvalue
if (isGenerating) return
isGenerating = true
try {
let npath =
'u/' +
($userStore?.username ?? $userStore?.email)?.split('@')[0] +
'/secret_arg/' +
generateRandomString(12)
let nvalue = '$var:' + npath
await VariableService.createVariable({
workspace: $workspaceStore!,
requestBody: {
value: password,
is_secret: true,
path: npath,
description: 'Ephemeral secret variable',
expires_at: new Date(Date.now() + 1000 * 60 * 60 * 24 * 7).toISOString()
}
})
path = npath
value = nvalue
debouncedUpdate()
} finally {
isGenerating = false
}
}
async function updateValue() {
await VariableService.updateVariable({
workspace: $workspaceStore!,
path: path,
requestBody: {
value: password
}
})
try {
await VariableService.updateVariable({
workspace: $workspaceStore!,
path: path,
requestBody: {
value: password
}
})
} catch (e) {
generateValue()
}
}
let timeout: NodeJS.Timeout | undefined = undefined
function debouncedUpdate() {
timeout && clearTimeout(timeout)
setTimeout(updateValue, 500)
timeout = setTimeout(updateValue, 500)
}
$: password && debouncedUpdate()
@@ -299,7 +299,7 @@
let storage: string | undefined = undefined
export async function open(
preSelectedFileKey: { s3: string; storage: string | undefined } | undefined = undefined
preSelectedFileKey: { s3: string; storage?: string } | undefined = undefined
) {
storage = preSelectedFileKey?.storage
if (preSelectedFileKey !== undefined) {
@@ -23,6 +23,7 @@
import { isCloudHosted } from '$lib/cloud'
import InstanceNameEditor from './InstanceNameEditor.svelte'
import Toggle from './Toggle.svelte'
import { instanceSettingsSelectedTab } from '$lib/stores';
let drawer: Drawer
let filter = ''
@@ -56,6 +57,9 @@
let tab: 'users' | string = 'users'
$: $instanceSettingsSelectedTab, tab = $instanceSettingsSelectedTab
$: tab, instanceSettingsSelectedTab.set(tab)
let nbDisplayed = 50
let instanceSettings
@@ -18,6 +18,9 @@
import ResolveConfig from '../helpers/ResolveConfig.svelte'
import ResolveStyle from '../helpers/ResolveStyle.svelte'
import { deepEqual } from 'fast-equals'
import { computeWorkspaceS3FileInputPolicy } from '../../editor/appUtilsS3'
import { defaultIfEmptyString } from '$lib/utils'
import { userStore } from '$lib/stores'
export let id: string
export let componentInput: AppInput | undefined
@@ -26,8 +29,16 @@
export let configuration: RichConfigurations
export let customCss: ComponentCustomCSS<'schemaformcomponent'> | undefined = undefined
const { worldStore, connectingInput, app, selectedComponent, componentControl } =
getContext<AppViewerContext>('AppViewerContext')
const {
worldStore,
connectingInput,
app,
selectedComponent,
componentControl,
appPath,
isEditor,
workspace
} = getContext<AppViewerContext>('AppViewerContext')
const iterContext = getContext<ListContext>('ListWrapperContext')
const listInputs: ListInputs | undefined = getContext<ListInputs>('ListInputs')
@@ -100,6 +111,14 @@
previousDefault = structuredClone(resolvedConfig.defaultValues)
args = previousDefault ?? {}
}
function computeS3ForceViewerPolicies() {
if (!isEditor) {
return undefined
}
const policy = computeWorkspaceS3FileInputPolicy()
return policy
}
</script>
{#each Object.keys(components['schemaformcomponent'].initialData.configuration) as key (key)}
@@ -140,6 +159,9 @@
bind:this={schemaForm}
displayType={Boolean(resolvedConfig.displayType)}
largeGap={Boolean(resolvedConfig.largeGap)}
appPath={defaultIfEmptyString(appPath, `u/${$userStore?.username ?? 'unknown'}/newapp`)}
{computeS3ForceViewerPolicies}
{workspace}
{css}
/>
</div>
@@ -26,6 +26,7 @@
import { get } from 'svelte/store'
import RefreshButton from '$lib/components/apps/components/helpers/RefreshButton.svelte'
import { ctxRegex } from '../../utils'
import { computeWorkspaceS3FileInputPolicy } from '../../editor/appUtilsS3'
// Component props
export let id: string
@@ -671,6 +672,14 @@
return undefined
}
}
function computeS3ForceViewerPolicies() {
if (!isEditor) {
return undefined
}
const policy = computeWorkspaceS3FileInputPolicy()
return policy
}
</script>
{#each Object.entries(fields ?? {}) as [key, v] (key)}
@@ -754,6 +763,9 @@
<div class="px-2 h-fit min-h-0">
<LightweightSchemaForm
schema={schemaStripped}
appPath={defaultIfEmptyString(appPath, `u/${$userStore?.username ?? 'unknown'}/newapp`)}
{computeS3ForceViewerPolicies}
{workspace}
bind:this={schemaForm}
bind:args
on:inputClicked={handleInputClick}
@@ -10,6 +10,9 @@
import { writable, type Writable } from 'svelte/store'
import FileUpload from '$lib/components/common/fileUpload/FileUpload.svelte'
import InitializeComponent from '../helpers/InitializeComponent.svelte'
import { computeS3FileInputPolicy } from '../../editor/appUtilsS3'
import { defaultIfEmptyString } from '$lib/utils'
import { userStore } from '$lib/stores'
export let id: string
export let configuration: RichConfigurations
@@ -77,6 +80,15 @@
{/if}
*/
let forceDisplayUploads: boolean = false
const { appPath, isEditor } = getContext<AppViewerContext>('AppViewerContext')
function computeForceViewerPolicies() {
if (!isEditor) {
return undefined
}
const policy = computeS3FileInputPolicy((configuration as any)?.type?.configuration?.s3, $app)
return policy
}
</script>
<InitializeComponent {id} />
@@ -139,5 +151,7 @@
outputs.result.set(value)
}}
{forceDisplayUploads}
appPath={defaultIfEmptyString(appPath, `u/${$userStore?.username ?? 'unknown'}/newapp`)}
{computeForceViewerPolicies}
/>
{/if}
@@ -87,6 +87,7 @@
import ToggleEnable from '$lib/components/common/toggleButton-v2/ToggleEnable.svelte'
import HideButton from './settingsPanel/HideButton.svelte'
import DeployOverrideConfirmationModal from '$lib/components/common/confirmationModal/DeployOverrideConfirmationModal.svelte'
import { computeS3FileInputPolicy, computeWorkspaceS3FileInputPolicy } from './appUtilsS3'
async function hash(message) {
try {
@@ -195,8 +196,12 @@
}
async function computeTriggerables() {
const items = allItems($app.grid, $app.subgrids)
console.log('items', items)
const allTriggers: ([string, TriggerableV2] | undefined)[] = (await Promise.all(
allItems($app.grid, $app.subgrids)
items
.flatMap((x) => {
let c = x.data as AppComponent
let r: { input: AppInput | undefined; id: string }[] = [
@@ -296,6 +301,44 @@
allTriggers.filter(Boolean) as [string, TriggerableV2][]
)
policy.triggerables_v2 = ntriggerables
const s3_inputs = items
.filter((x) => (x.data as AppComponent).type === 's3fileinputcomponent')
.map((x) => {
const c = x.data as AppComponent
const config = c.configuration as any
return computeS3FileInputPolicy(config?.type?.configuration?.s3, $app)
})
.filter(Boolean) as {
allowed_resources: string[]
allow_user_resources: boolean
file_key_regex: string
}[]
if (
items.findIndex((x) => {
const c = x.data as AppComponent
if (c.type === 'schemaformcomponent') {
return (
Object.values((c.componentInput as any)?.value?.properties ?? {}).findIndex(
(p: any) => p?.type === 'object' && p?.format === 'resource-s3_object'
) !== -1
)
} else if (c.type === 'formbuttoncomponent' || c.type === 'formcomponent') {
return (
Object.values((c.componentInput as any)?.fields ?? {}).findIndex(
(p: any) => p?.fieldType === 'object' && p?.format === 'resource-s3_object'
) !== -1
)
} else {
return false
}
}) !== -1
) {
s3_inputs.push(computeWorkspaceS3FileInputPolicy())
}
policy.s3_inputs = s3_inputs
}
async function processRunnable(
@@ -0,0 +1,82 @@
import type { AppInput, EvalInputV2 } from '../inputType'
import type { App } from '../types'
import { collectOneOfFields } from './appUtils'
function filenameExprToRegex(template: string) {
const filenameEscaped = template.replaceAll('${file.name}', '<file_name>') // replace filename with placeholder
const escapedTemplate = filenameEscaped
.slice(1, -1) // remove quotes
.replace(/[.*+?^${}()|[\]\\]/g, '\\$&') // escape regex special characters
const regexPattern = escapedTemplate.replaceAll('<file_name>', '[^/]+') // replace filename placeholder with regex pattern
return `^${regexPattern}$`
}
function staticToRegex(str: string) {
return `^${str.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}$`
}
function checkIfExprIsString(input: string) {
return /^(['"`])[^'"`]*\1$/g.test(input)
}
function checkIfEvalIsStringWithFilename(input: EvalInputV2) {
if (input.connections.length > 0) {
return false
} else {
return checkIfExprIsString(input.expr.replaceAll('${file.name}', ''))
}
}
function removeResourcePrefix(resource: string) {
return resource.replace(/^\$res:/, '')
}
export function computeWorkspaceS3FileInputPolicy() {
return {
allow_workspace_resource: true,
allowed_resources: [],
allow_user_resources: false,
file_key_regex: ''
}
}
export function computeS3FileInputPolicy(s3Config: any, app: App) {
const resourceInput = s3Config?.resource as AppInput | undefined
const pathTemplateInput = s3Config?.pathTemplate as AppInput | undefined
const allow_workspace_resource =
!resourceInput || (resourceInput.type === 'static' && !resourceInput.value)
const allowed_resources: string[] = resourceInput
? resourceInput.type === 'static'
? resourceInput.value
? [removeResourcePrefix(resourceInput.value)]
: []
: collectOneOfFields(
{
s3_resource: resourceInput
},
app
).s3_resource?.map((s) => removeResourcePrefix(s)) ?? []
: []
const allow_user_resources =
(resourceInput?.type === 'evalv2' && resourceInput?.allowUserResources) ?? false
let file_key_regex = '^.*$'
if (pathTemplateInput) {
if (pathTemplateInput.type === 'static') {
file_key_regex = staticToRegex(pathTemplateInput.value)
} else if (
pathTemplateInput.type === 'evalv2' &&
checkIfEvalIsStringWithFilename(pathTemplateInput)
) {
file_key_regex = filenameExprToRegex(pathTemplateInput.expr)
}
}
return {
allow_workspace_resource,
allowed_resources,
allow_user_resources,
file_key_regex
}
}
@@ -195,7 +195,7 @@
{:else if componentInput?.type === 'user'}
<span class="text-2xs italic text-tertiary">Field's value is set by the user</span>
{/if}
{#if (componentInput?.type === 'evalv2' || componentInput?.type === 'connected' || componentInput?.type === 'user') && fieldType == 'object' && format?.startsWith('resource-')}
{#if (componentInput?.type === 'evalv2' || componentInput?.type === 'connected' || componentInput?.type === 'user') && ((fieldType == 'object' && format?.startsWith('resource-') && format !== 'resource-s3_object') || fieldType == 'resource')}
<div class="flex flex-row items-center">
<Toggle
size="xs"
@@ -335,7 +335,15 @@
}
</script>
<svelte:window on:keydown={handleKeyDown} on:keyup={handleKeyUp} />
<svelte:window
on:focus={() => {
if ($isCtrlOrMetaPressedStore) {
$isCtrlOrMetaPressedStore = false
}
}}
on:keydown={handleKeyDown}
on:keyup={handleKeyUp}
/>
<div
class="svlt-grid-container"
@@ -4,6 +4,7 @@
import Button from '../../common/button/Button.svelte'
import { twMerge } from 'tailwind-merge'
import type { ReadFileAs } from './model'
import { sendUserToast } from '$lib/toast'
type ConvertedFile = string | ArrayBuffer | null
@@ -89,7 +90,12 @@
event.preventDefault()
if (event.dataTransfer) {
if (event.dataTransfer.files && event.dataTransfer.files.length) {
onChange(event.dataTransfer.files)
if (!multiple && event.dataTransfer.files.length > 1) {
sendUserToast('Only one file can be uploaded at a time')
return
} else {
onChange(event.dataTransfer.files)
}
}
}
}
@@ -4,7 +4,7 @@
import Button from '$lib/components/common/button/Button.svelte'
import { sendUserToast } from '$lib/toast'
import { workspaceStore } from '$lib/stores'
import { userStore, workspaceStore } from '$lib/stores'
import { HelpersService } from '$lib/gen'
import { writable, type Writable } from 'svelte/store'
import { Ban, CheckCheck, FileWarning, Files, RefreshCcw, Trash } from 'lucide-svelte'
@@ -14,7 +14,9 @@
export let acceptedFileTypes: string[] | undefined = ['*']
export let allowMultiple: boolean = true
export let containerText: string = 'Drag and drop files here or click to browse'
export let containerText: string = allowMultiple
? 'Drag and drop files here or click to browse'
: 'Drag and drop a file here or click to browse'
export let customResourcePath: string | undefined = undefined
export let customResourceType: 's3' | 'azure_blob' | undefined = undefined // when customResourcePath is provided, this should be provided as well. Will default to S3 if not
export let customClass: string = ''
@@ -25,6 +27,17 @@
export let defaultValue: string | undefined = undefined
export let workspace: string | undefined = undefined
export let fileUploads: Writable<FileUploadData[]> = writable([])
export let appPath: string | undefined = undefined
export let computeForceViewerPolicies:
| (() =>
| {
allowed_resources: string[]
allow_user_resources: boolean
allow_workspace_resource: boolean
file_key_regex: string
}
| undefined)
| undefined = undefined
const dispatch = createEventDispatcher()
@@ -109,6 +122,30 @@
if (fileExtension) {
params.append('file_extension', fileExtension)
}
if (fileToUpload.type) {
params.append('content_type', fileToUpload.type)
}
if (computeForceViewerPolicies !== undefined) {
const forceViewerPolicies = computeForceViewerPolicies()
if (forceViewerPolicies) {
params.append(
'force_viewer_allowed_resources',
forceViewerPolicies.allowed_resources.join(',')
)
params.append(
'force_viewer_allow_user_resources',
JSON.stringify(forceViewerPolicies.allow_user_resources)
)
params.append(
'force_viewer_allow_workspace_resource',
JSON.stringify(forceViewerPolicies.allow_workspace_resource)
)
params.append('force_viewer_file_key_regex', forceViewerPolicies.file_key_regex)
}
}
// let response = await fetch(
// `/api/w/${$workspaceStore}/job_helpers/multipart_upload_s3_file?${params.toString()}`,
// {
@@ -152,9 +189,16 @@
}
}
})
xhr?.open(
'POST',
`/api/w/${workspace ?? $workspaceStore}/job_helpers/upload_s3_file?${params.toString()}`,
appPath
? `/api/w/${
workspace ?? $workspaceStore
}/apps_u/upload_s3_file/${appPath}?${params.toString()}`
: `/api/w/${
workspace ?? $workspaceStore
}/job_helpers/upload_s3_file?${params.toString()}`,
true
)
xhr?.setRequestHeader('Content-Type', 'application/octet-stream')
@@ -303,7 +347,7 @@
</Button>
{/if}
{#if fileUpload.progress === 100 && !fileUpload.cancelled}
{#if fileUpload.progress === 100 && !fileUpload.cancelled && $userStore}
<Button
size="xs2"
color="red"
@@ -400,7 +444,6 @@
accept={acceptedFileTypes?.join(',')}
multiple={allowMultiple}
returnFileNames
includeMimeType
on:change={({ detail }) => {
forceDisplayUploads = false
handleChange(detail)
@@ -78,6 +78,10 @@
newItem={newFlow}
isFlow={true}
/>
{:else if $selectedId.startsWith('subflow:')}
<div class="p-4"
>Selected step is witin an expanded subflow and is not directly editable in the flow editor</div
>
{:else}
{@const dup = checkDup($flowStore.value.modules)}
{#if dup}
@@ -307,7 +307,7 @@
{/if}
{#if inlineScripts?.length > 0}
<div class="pb-0 flex flex-row items-center gap-2">
<div class="pb-0 flex flex-row items-center gap-2 -mt-[3px]">
<div class=" text-2xs font-light text-secondary ml-2"
>New {selectedKind != 'script' ? selectedKind + ' ' : ''}script</div
>
@@ -130,8 +130,7 @@
}}
/>
</div>
<div
>``
<div>
<div class="mb-2 text-sm font-bold">Run in parallel</div>
<Toggle
bind:checked={mod.value.parallel}
@@ -253,6 +253,10 @@
forceReload++
await reload(flowModule)
}
if (flowModule.value.type == 'flow') {
forceReload++
await reload(flowModule)
}
}}
on:createScriptFromInlineScript={async () => {
const [module, state] = await createScriptFromInlineScript(
@@ -361,7 +365,9 @@
</div>
{/if}
{:else if flowModule.value.type === 'flow'}
<FlowPathViewer path={flowModule.value.path} />
{#key forceReload}
<FlowPathViewer path={flowModule.value.path} />
{/key}
{/if}
</Pane>
<Pane bind:size={editorSettingsPanelSize} minSize={20}>
@@ -378,13 +384,14 @@
class={advancedSelected === 'runtime' ? 'h-[calc(100%-68px)]' : 'h-[calc(100%-34px)]'}
>
{#if selected === 'inputs' && (flowModule.value.type == 'rawscript' || flowModule.value.type == 'script' || flowModule.value.type == 'flow')}
<div class="h-full overflow-auto px-2 bg-surface" id="flow-editor-step-input">
<div class="h-full overflow-auto bg-surface" id="flow-editor-step-input">
<PropPickerWrapper
pickableProperties={stepPropPicker.pickableProperties}
error={failureModule}
noPadding
>
<InputTransformSchemaForm
class="px-1 xl:px-2"
bind:this={inputTransformSchemaForm}
pickableProperties={stepPropPicker.pickableProperties}
schema={$flowStateStore[$selectedId]?.schema ?? {}}
@@ -5,10 +5,12 @@
import {
Bed,
Database,
ExternalLink,
Gauge,
GitFork,
Pen,
PhoneIncoming,
RefreshCcw,
Repeat,
Save,
Square,
@@ -139,6 +141,32 @@
</Button>
{/if}
{/if}
{#if module.value.type === 'flow'}
<Button
size="xs"
color="light"
on:click={async () => {
if (module.value.type == 'flow') {
window.open(`/flows/edit/${module.value.path}`, '_blank', 'noopener,noreferrer')
}
}}
startIcon={{ icon: Pen }}
iconOnly={false}
>
Edit <ExternalLink size={12} />
</Button>
<Button
size="xs"
color="light"
on:click={async () => {
dispatch('reload')
}}
startIcon={{
icon: RefreshCcw
}}
iconOnly={true}
/>
{/if}
<div class="px-0.5" />
{#if module.value.type === 'rawscript'}
<FlowModuleWorkerTagSelect bind:tag={module.value.tag} />
@@ -57,6 +57,7 @@
{#if flowModule.sleep && schema.properties['sleep']}
<div class="border">
<PropPickerWrapper
noFlowPlugConnect={true}
notSelectable
{result}
displayContext={false}
@@ -72,6 +73,7 @@
{schema}
{previousModuleId}
argExtra={{ seconds: true }}
bind:editor
/>
</PropPickerWrapper>
</div>
@@ -187,6 +187,7 @@
<div class="border">
<PropPickerWrapper
{result}
noFlowPlugConnect
displayContext={false}
pickableProperties={undefined}
on:select={({ detail }) => {
@@ -200,6 +201,7 @@
argName="groups"
{schema}
{previousModuleId}
bind:editor
/>
</PropPickerWrapper>
</div>
@@ -276,8 +276,13 @@
</Popover>
<div class="flex items-center space-x-2 relative max-w-[25%]" bind:clientWidth={idBadgeWidth}>
{#if id && id !== 'preprocessor' && !id.startsWith('failure')}
<Badge color="indigo" wrapperClass="max-w-full" baseClass="max-w-full truncate" title={id}>
{#if id && id !== 'preprocessor' && !id.startsWith('failure') && !id.startsWith('subflow:')}
<Badge
color="indigo"
wrapperClass="max-w-full"
baseClass="max-w-full truncate !px-1"
title={id}
>
<span class="max-w-full text-2xs truncate">{id}</span></Badge
>
{#if deletable}
@@ -290,6 +295,10 @@ hover:border-blue-700 hover:!visible {hover ? '' : '!hidden'}"
title="Edit Id"><Pencil size={14} /></button
>
{/if}
{:else if id?.startsWith('subflow:')}
<Badge color="blue" wrapperClass="max-w-full" baseClass="!px-1" title={id}>
<span class="max-w-full text-2xs truncate">{id.substring('subflow:'.length)}</span></Badge
>
{/if}
</div>
</div>
@@ -26,7 +26,7 @@
import { getDependentComponents } from '../flowExplorer'
import type { FlowCopilotContext } from '$lib/components/copilot/flow'
import { fade } from 'svelte/transition'
import { copilotInfo, tutorialsToDo } from '$lib/stores'
import { copilotInfo, tutorialsToDo, workspaceStore } from '$lib/stores'
import FlowTutorials from '$lib/components/FlowTutorials.svelte'
import { ignoredTutorials } from '$lib/components/tutorials/ignoredTutorials'
@@ -44,6 +44,7 @@
export let disableSettings = false
export let newFlow: boolean = false
export let smallErrorHandler = false
export let workspace: string | undefined = $workspaceStore
let flowTutorials: FlowTutorials | undefined = undefined
@@ -345,6 +346,7 @@
preprocessorModule={$flowStore.value?.preprocessor_module}
{selectedId}
{flowInputsStore}
{workspace}
on:delete={({ detail }) => {
let e = detail.detail
dependents = getDependentComponents(e.id, $flowStore)
@@ -20,7 +20,6 @@
export let modules: FlowModule[]
export let moving: string | undefined = undefined
export let duration_ms: number | undefined = undefined
export let isTrigger: boolean = false
export let retries: number | undefined = undefined
export let flowJobs:
@@ -54,7 +54,7 @@
<div class="truncate text-2xs text-center"><pre>{preLabel}</pre></div>
{/if}
</div>
{#if id && !hideId}
{#if id && !hideId && !id?.startsWith('subflow:')}
<div class="flex items-center shrink min-w-0">
<Badge color="indigo" wrapperClass="w-full" baseClass="max-w-full" title={id}>
<span class="max-w-full text-2xs truncate">{id}</span>
@@ -74,7 +74,10 @@ function getFlowInput(
if (parentState && parentModule) {
if (
parentState.previewArgs &&
!(parentModule.value?.type === 'forloopflow' && parentModule.value?.modules?.length === 1)
!(parentModule.value?.type === 'forloopflow' && parentModule.value?.modules?.length === 1) &&
parentState &&
typeof parentState.previewArgs == 'object' &&
`iter` in parentState.previewArgs
) {
return { ...topFlowInput, ...parentState.previewArgs }
} else {
@@ -283,11 +286,8 @@ export function filterNestedObject(obj: any, nestedKeys: string[]) {
if (nestedKeys[0] === '') {
return obj
}
const regexes = buildPrefixRegex(Object.keys(obj))
const matches = regexes.filter(({ regex }) => regex.test(nestedKeys[0]))
return Object.fromEntries(
Object.entries(obj).filter(([key]) => matches.some(({ word }) => word === key))
)
return Object.fromEntries(Object.entries(obj).filter(([key]) => key.startsWith(nestedKeys[0])))
}
const [key, ...rest] = nestedKeys
if (obj && typeof obj === 'object' && key in obj) {
@@ -43,6 +43,7 @@
export let noPadding: boolean = false
export let alwaysOn: boolean = false
export let paneClass: string = ''
export let noFlowPlugConnect = false
const propPickerConfig: Writable<PropPickerConfig | undefined> = writable<
PropPickerConfig | undefined
@@ -67,13 +68,15 @@
}
}
propPickerConfig.set(config)
flowPropPickerConfig.set({
...config,
clearFocus: () => {
propPickerConfig.set(undefined)
flowPropPickerConfig.set(undefined)
}
})
if (!noFlowPlugConnect) {
flowPropPickerConfig.set({
...config,
clearFocus: () => {
propPickerConfig.set(undefined)
flowPropPickerConfig.set(undefined)
}
})
}
},
clearFocus: () => {
flowPropPickerConfig.set(undefined)
@@ -121,7 +124,7 @@
: 'transparent'}
animationDuration="4s"
>
{#if result}
{#if result != undefined}
<PropPickerResult
{result}
{extraResults}
@@ -139,6 +142,7 @@
{alwaysOn}
{displayContext}
{error}
previousId={pickableProperties?.previousId}
{pickableProperties}
allowCopy={!notSelectable && !$propPickerConfig}
on:select={({ detail }) => {

Some files were not shown because too many files have changed in this diff Show More