Compare commits

..

3 Commits

Author SHA1 Message Date
Ruben Fiszel
2f46459844 all 2024-04-14 19:37:12 +02:00
Ruben Fiszel
f82dfa7e97 all 2024-04-14 19:37:03 +02:00
Ruben Fiszel
a605d79f21 all 2024-04-14 19:24:54 +02:00
421 changed files with 7768 additions and 16248 deletions

View File

@@ -16,7 +16,6 @@ sed -i '' -e "/\"version\": /s/: .*,/: \"$VERSION\",/" ${root_dirpath}/frontend/
sed -i '' -e "/^version =/s/= .*/= \"$VERSION\"/" ${root_dirpath}/python-client/wmill/pyproject.toml
sed -i '' -e "/^windmill-api =/s/= .*/= \"\\^$VERSION\"/" ${root_dirpath}/python-client/wmill/pyproject.toml
sed -i '' -e "/^version =/s/= .*/= \"$VERSION\"/" ${root_dirpath}/python-client/wmill_pg/pyproject.toml
sed -i '' -e "/^ModuleVersion =/s/= .*/= '$VERSION'/" ${root_dirpath}/powershell-client/WindmillClient/WindmillClient.psd1
# sed -i '' -e "/^wmill =/s/= .*/= \"\\^$VERSION\"/" python-client/wmill_pg/pyproject.toml
sed -i '' -e "/^wmill =/s/= .*/= \">=$VERSION\"/" ${root_dirpath}/lsp/Pipfile
sed -i '' -e "/^wmill_pg =/s/= .*/= \">=$VERSION\"/" ${root_dirpath}/lsp/Pipfile

View File

@@ -12,12 +12,10 @@ sed -i -e "/^export const VERSION =/s/= .*/= \"v$VERSION\";/" ${root_dirpath}/be
sed -i -e "/version: /s/: .*/: $VERSION/" ${root_dirpath}/backend/windmill-api/openapi.yaml
sed -i -e "/version: /s/: .*/: $VERSION/" ${root_dirpath}/openflow.openapi.yaml
sed -i -e "/\"version\": /s/: .*,/: \"$VERSION\",/" ${root_dirpath}/typescript-client/package.json
sed -i -e "/\"version\": /s/: .*,/: \"$VERSION\",/" ${root_dirpath}/typescript-client/jsr.json
sed -i -e "/\"version\": /s/: .*,/: \"$VERSION\",/" ${root_dirpath}/frontend/package.json
sed -i -e "/^version =/s/= .*/= \"$VERSION\"/" ${root_dirpath}/python-client/wmill/pyproject.toml
sed -i -e "/^windmill-api =/s/= .*/= \"\\^$VERSION\"/" ${root_dirpath}/python-client/wmill/pyproject.toml
sed -i -e "/^version =/s/= .*/= \"$VERSION\"/" ${root_dirpath}/python-client/wmill_pg/pyproject.toml
sed -i -e "/^ModuleVersion =/s/= .*/= '$VERSION'/" ${root_dirpath}/powershell-client/WindmillClient/WindmillClient.psd1
# sed -i -e "/^wmill =/s/= .*/= \"\\^$VERSION\"/" ${root_dirpath}/python-client/wmill_pg/pyproject.toml
sed -i -e "/^wmill =/s/= .*/= \">=$VERSION\"/" ${root_dirpath}/lsp/Pipfile
sed -i -e "/^wmill_pg =/s/= .*/= \">=$VERSION\"/" ${root_dirpath}/lsp/Pipfile

View File

@@ -136,7 +136,7 @@ jobs:
platforms: linux/amd64,linux/arm64
push: true
build-args: |
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect
tags: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee:dev
${{ steps.meta-ee-public.outputs.tags }}
@@ -199,7 +199,7 @@ jobs:
platforms: linux/amd64
push: true
build-args: |
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect
PYTHON_IMAGE=python:3.12.2-slim-bookworm
tags: |
${{ steps.meta-ee-public-py312.outputs.tags }}

View File

@@ -1,16 +0,0 @@
name: Publish powershell-client
on:
push:
tags:
- "v*"
workflow_dispatch:
jobs:
publish_gallery:
runs-on: ubicloud-standard-8
steps:
- uses: actions/checkout@v4
- run: . ./powershell-client/publish.ps1
shell: pwsh
env:
NUGET_API_KEY: ${{ secrets.NUGET_API_KEY }}

View File

@@ -1,16 +0,0 @@
name: Publish typescript-client on JSR
on:
push:
tags:
- "v*"
jobs:
publish:
runs-on: ubuntu-latest
permissions:
contents: read
id-token: ${{ secrets.JSR_OIDC_ID_TOKEN }}
steps:
- uses: actions/checkout@v4
- run: cd typescript-client && ./publish.jsr.sh

File diff suppressed because it is too large Load Diff

View File

@@ -12,14 +12,12 @@ RUN apt-get -y update \
RUN rustup component add rustfmt
RUN CARGO_NET_GIT_FETCH_WITH_CLI=true cargo install cargo-chef --version ^0.1
RUN cargo install sccache --version ^0.8
ENV RUSTC_WRAPPER=sccache SCCACHE_DIR=/backend/sccache
RUN CARGO_NET_GIT_FETCH_WITH_CLI=true cargo install cargo-chef
WORKDIR /windmill
ENV SQLX_OFFLINE=true
# ENV CARGO_INCREMENTAL=1
ENV CARGO_INCREMENTAL=1
FROM node:20-alpine as frontend
@@ -48,9 +46,7 @@ FROM rust_base AS planner
COPY ./openflow.openapi.yaml /openflow.openapi.yaml
COPY ./backend ./
RUN --mount=type=cache,target=/usr/local/cargo/registry \
--mount=type=cache,target=$SCCACHE_DIR,sharing=locked \
CARGO_NET_GIT_FETCH_WITH_CLI=true cargo chef prepare --recipe-path recipe.json
RUN CARGO_NET_GIT_FETCH_WITH_CLI=true cargo chef prepare --recipe-path recipe.json
FROM rust_base AS builder
ARG features=""
@@ -59,9 +55,7 @@ COPY --from=planner /windmill/recipe.json recipe.json
RUN apt-get update && apt-get install -y libxml2-dev libxmlsec1-dev clang libclang-dev cmake
RUN --mount=type=cache,target=/usr/local/cargo/registry \
--mount=type=cache,target=$SCCACHE_DIR,sharing=locked \
CARGO_NET_GIT_FETCH_WITH_CLI=true RUST_BACKTRACE=1 cargo chef cook --release --features "$features" --recipe-path recipe.json
RUN CARGO_NET_GIT_FETCH_WITH_CLI=true RUST_BACKTRACE=1 cargo chef cook --release --features "$features" --recipe-path recipe.json
COPY ./openflow.openapi.yaml /openflow.openapi.yaml
COPY ./backend ./
@@ -70,9 +64,7 @@ COPY --from=frontend /frontend /frontend
COPY --from=frontend /backend/windmill-api/openapi-deref.yaml ./windmill-api/openapi-deref.yaml
COPY .git/ .git/
RUN --mount=type=cache,target=/usr/local/cargo/registry \
--mount=type=cache,target=$SCCACHE_DIR,sharing=locked \
CARGO_NET_GIT_FETCH_WITH_CLI=true cargo build --release --features "$features"
RUN CARGO_NET_GIT_FETCH_WITH_CLI=true cargo build --release --features "$features"
FROM ${DEBIAN_IMAGE} as downloader
@@ -171,7 +163,7 @@ COPY --from=builder /windmill/target/release/windmill ${APP}/windmill
COPY --from=downloader --chmod=755 /deno /usr/bin/deno
COPY --from=oven/bun:1.1.7 /usr/local/bin/bun /usr/bin/bun
COPY --from=oven/bun:1.1.0 /usr/local/bin/bun /usr/bin/bun
# add the docker client to call docker from a worker if enabled
COPY --from=docker:dind /usr/local/bin/docker /usr/local/bin/

View File

@@ -13,7 +13,7 @@ any snippets of code that require a positive license check to be activated.
Those snippets and files are under a proprietary and commercial license. Private
and public forks MUST not include any of the above proprietary and commercial
code. Windmill Labs, Inc. provide tools to clean the codebase from those
snippets upon demand. The files under python-client/ deno-client/ go-client/ powershell-client/ are
snippets upon demand. The files under python-client/ deno-client/ go-client/ are
Apache 2.0 Licensed.
The openapi files, including the OpenFlow spec is Apache 2.0 Licensed.

View File

@@ -352,44 +352,6 @@ you to have it being synced automatically everyday.
| TIMEOUT_WAIT_RESULT | 20 | The number of seconds to wait before timeout on the 'run_wait_result' endpoint | Worker |
| QUEUE_LIMIT_WAIT_RESULT | None | The number of max jobs in the queue before rejecting immediately the request in 'run_wait_result' endpoint. Takes precedence on the query arg. If none is specified, there are no limit. | Worker |
| DENO_AUTH_TOKENS | None | Custom DENO_AUTH_TOKENS to pass to worker to allow the use of private modules | Worker |
| DISABLE_RESPONSE_LOGS | false | Disable response logs | Server |
## Run a local dev setup
### only Frontend
This will use the backend of <https://app.windmill.dev> but your own frontend
with hot-code reloading.
1. Install [caddy](https://caddyserver.com)
2. Go to `frontend/`:
1. `npm install`, `npm run generate-backend-client` then `npm run dev`
2. In another shell `sudo caddy run --config CaddyfileRemote`
3. Et voilà, windmill should be available at `http://localhost/`
### Backend + Frontend
See the [./frontend/README_DEV.md](./frontend/README_DEV.md) file for all
running options.
1. Create a Postgres Database for Windmill and create an admin role inside your
Postgres setup.
The easiest way to get a working db is to run
```
cargo install sqlx-cli
env DATABASE_URL=<YOUR_DATABASE_URL> sqlx migrate run
```
This will also avoid compile time issue with sqlx's `query!` macro
2. Install [nsjail](https://github.com/google/nsjail) and have it accessible in
your PATH
3. Install deno and python3, have the bins at `/usr/bin/deno` and
`/usr/local/bin/python3`
4. Install [caddy](https://caddyserver.com)
5. Install the [lld linker](https://lld.llvm.org/)
6. Go to `frontend/`:
1. `npm install`, `npm run generate-backend-client` then `npm run dev`
2. You might need to set some extra heap space for the node runtime `export NODE_OPTIONS="--max-old-space-size=4096"`
3. In another shell `npm run build` otherwise the backend will not find the `frontend/build` folder and will not compile.
4. In another shell `sudo caddy run --config Caddyfile`
7. Go to `backend/`:
`env DATABASE_URL=<DATABASE_URL_TO_YOUR_WINDMILL_DB> RUST_LOG=info cargo run`
8. Et voilà, windmill should be available at `http://localhost/`
## Contributors

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30)",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28)",
"describe": {
"columns": [],
"parameters": {
@@ -66,12 +66,10 @@
"Bool",
"Int4",
"Varchar",
"Bool",
"Bool",
"Varchar"
"Bool"
]
},
"nullable": []
},
"hash": "8e7ff45c5378c3a3406ba94dc0653afa5d28c072203617c598caefaaf1bafcfb"
"hash": "020d33ed5d47350b456783fd548422ea8dcf2d786d0e9fa849754db82c9fa378"
}

View File

@@ -1,20 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT created_at FROM metrics WHERE id = 'author_count' ORDER BY created_at DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": []
},
"nullable": [
false
]
},
"hash": "021be0f26ea87e587e656b24a9a94538efbf54a1447a3898e19773789cfc9063"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, running, is_flow_step FROM queue WHERE scheduled_for < now() AND workspace_id = $1 AND schedule_path IS NULL",
"query": "UPDATE queue SET canceled = true, canceled_by = $2, scheduled_for = now(), suspend = 0 WHERE scheduled_for < now() AND workspace_id = $1 AND schedule_path IS NULL RETURNING id, running, is_flow_step",
"describe": {
"columns": [
{
@@ -21,7 +21,8 @@
],
"parameters": {
"Left": [
"Text"
"Text",
"Varchar"
]
},
"nullable": [
@@ -30,5 +31,5 @@
true
]
},
"hash": "caeb49629b8673c1f1c84a6e40c3e2d2c3bc3fdbde530a0a6b6fd68a22b867c3"
"hash": "18699cb0eca25b6bde05d81571dfdea8cafd0043634f61b0f652a93767c9c30a"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM workspace_invite WHERE workspace_id = $1 AND email = $2",
"query": "DELETE FROM folder WHERE name = $1 AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
@@ -11,5 +11,5 @@
},
"nullable": []
},
"hash": "0d7ba88a9810e434aa00fd63bbf416cbe222f2c67ccc8aa92e651c2bea4c2d7b"
"hash": "26e4ec75366d1e46a98710f29066b40e66a802f98eeabbb3ae5bebe3aeb6b3f8"
}

View File

@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM metrics \n WHERE (id = 'author_count' OR id = 'operator_count' OR id = 'worker_usage') AND created_at < NOW() - INTERVAL '6 month'",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "3667d72d23e8c35ab96d5e5d1fbfd94a8a9f74b6398e21f13c206c3f9427c6a9"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT tag, count(*) as count FROM queue WHERE\n scheduled_for <= now() - ('3 seconds')::interval AND running = false\n GROUP BY tag",
"query": "SELECT tag, count(*) as count FROM queue WHERE\n scheduled_for <= now() - ('3 seconds')::interval AND running = false\n GROUP BY tag",
"describe": {
"columns": [
{
@@ -22,5 +22,5 @@
null
]
},
"hash": "02b516dac764662194db1bc33e365c01f40bae70af3683f1f09748f6020f0d49"
"hash": "3e0e5e0076ae0f7771abd64f8c5bb7e003e3f85351c946a012037e0c874acf1a"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "4613382f7b031a2b667f86d9af995065a99a63c407fd58aa293fb269e032121d"
}

View File

@@ -1,17 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $4",
"query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2 WHERE worker = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int4",
"TextArray",
"Float4",
"Text"
]
},
"nullable": []
},
"hash": "54fef88cc6b9e8db7c07fccbaa845edfefa339153a32e468faad9f063008863d"
"hash": "47beea5cd6324b53bfb349665fb215280f32b70a617fde87f70ea53ca9ade39f"
}

View File

@@ -1,38 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT worker, worker_instance, vcpus, memory FROM worker_ping WHERE ping_at > NOW() - INTERVAL '2 minutes'",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "worker",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "worker_instance",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "vcpus",
"type_info": "Int8"
},
{
"ordinal": 3,
"name": "memory",
"type_info": "Int8"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
true,
true
]
},
"hash": "4887db074e9058bc0c1887428d6b2a897fe4b30c83f2aa5e9464c156f08f1c3a"
}

View File

@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO metrics (id, value) VALUES ('worker_usage', $1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Jsonb"
]
},
"nullable": []
},
"hash": "5ce9a9e0669b299b998bc9080d24f8d6f56eb049bec2e0745d27ca3a4fc0c296"
}

View File

@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM schedule WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "61e9662fe42506131222412ab3de48cf6485dea10aa3a2f97c0fd6322a0cb17f"
}

View File

@@ -5,7 +5,7 @@
"columns": [
{
"ordinal": 0,
"name": "bool",
"name": "?column?",
"type_info": "Bool"
}
],

View File

@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version, vcpus, memory) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"TextArray",
"Varchar",
"Varchar",
"Varchar",
"Int8",
"Int8"
]
},
"nullable": []
},
"hash": "6afc5c7cbb3abe11ade0cedf1f7328005ce4de3165cdd998e5a0d27e044c7153"
}

View File

@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "6f12be65a4fe3eb39292164363f557de9cef7017dcfbcd40370b849a288c52e3"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS (SELECT 1 FROM usr WHERE workspace_id = $1 AND email = $2)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "7142222bee1f60bc56752e377b96ac6e4e981ea219c173146088faf1ad9f3822"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM folder WHERE name = $1 AND workspace_id = $2 RETURNING 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "748904c35cdbb6c7b8a8e0024b341278bf2bb727f2fe0427847565fb9c774abc"
}

View File

@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"TextArray",
"Varchar",
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "7b609ed87f974dfe887778746230b40b485658a75a1186c6ebb4c2de1b52a2fa"
}

View File

@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH queue_metrics as (\n SELECT id, value, created_at\n FROM metrics\n WHERE id LIKE 'queue_%'\n AND created_at > now() - interval '14 day'\n ORDER BY created_at ASC\n )\n SELECT id, array_agg(json_build_object('value', value, 'created_at', created_at)) as \"values!\"\n FROM queue_metrics\n GROUP BY id\n ORDER BY id ASC",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "values!",
"type_info": "JsonArray"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
null
]
},
"hash": "7c95f3652de4561b18c1a340db3e0113a4777d38de4468fc41b22ce5b6fd1db7"
}

View File

@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM metrics WHERE id LIKE 'queue_%' AND created_at < NOW() - INTERVAL '14 day'",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "7c9a464ac807051b99fe37f2078f1b17f824e6d9b1124db618855a15a98e31f6"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT usage.usage + 1 FROM usage \n WHERE is_workspace IS FALSE AND\n month_ = EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date)\n AND id = $1",
"query": "SELECT usage.usage + 1 FROM usage \n WHERE is_workspace IS FALSE AND\n month_ = EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date)\n AND id = $1",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
null
]
},
"hash": "94ff696b4d3904e3823ef637fa8f1f0d0bdac01040c81b31514326417eb58cee"
"hash": "7d93eb90163516718c85f28f8f05093133c5cbc96414000ac364583114d0ce77"
}

View File

@@ -1,20 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": []
},
"nullable": [
false
]
},
"hash": "82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92"
}

View File

@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO metrics (id, value) VALUES ($1, $2)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Jsonb"
]
},
"nullable": []
},
"hash": "8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15"
}

View File

@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON completed_job USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_label';",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "89e72f997e27b9298bd923e7f2b546c5ff061bd9ed63798024306456ca148aec"
}

View File

@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2 WHERE worker = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "8c62e3bb264c7336b3b3a68677993ba3b729c7358cc3da14513d8b447ce65aaf"
}

View File

@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO metrics (id, value)\n VALUES ($1, to_jsonb((SELECT EXTRACT(EPOCH FROM now() - scheduled_for)\n FROM queue WHERE tag = $2 AND running = false AND scheduled_for <= now() - ('3 seconds')::interval\n ORDER BY priority DESC NULLS LAST, scheduled_for, created_at LIMIT 1)))",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "9bf41c3161a02b7d0731c4e1d79519cef5255f5df1b759af3aa4985bb64313e5"
}

View File

@@ -1,16 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE((SELECT MIN(started_at) as min_started_at\n FROM queue\n WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $2 AND canceled = false\n GROUP BY script_path), $3) as min_started_at, now() AS now",
"query": "SELECT COALESCE((SELECT MIN(started_at) as min_started_at\n FROM queue\n WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $2 AND canceled = false\n GROUP BY script_path), $3)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "min_started_at",
"type_info": "Timestamptz"
},
{
"ordinal": 1,
"name": "now",
"name": "coalesce",
"type_info": "Timestamptz"
}
],
@@ -22,9 +17,8 @@
]
},
"nullable": [
null,
null
]
},
"hash": "3901cce744c9b246b661c817e068bdb3b1ab504ff8070fcccf6c909ad75f1f6d"
"hash": "abc7c72dfe9b01cde6f5b206300ed33e3d15b16bce2510160166a66fb7598e61"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM queue WHERE schedule_path = $1 AND running = false AND workspace_id = $2 AND is_flow_step = false",
"query": "DELETE FROM queue WHERE schedule_path = $1 AND running = false AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
@@ -11,5 +11,5 @@
},
"nullable": []
},
"hash": "8d4235984f27d8b939ffd5c660d5b57dc38dd3b2643361ed3b7cdcd1534d2e21"
"hash": "ade89de6e8527c543b182229f1febeb2513ad58b03ab526df148582264fb3a44"
}

View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate\n FROM worker_ping\n WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n ORDER BY ping_at desc LIMIT $2 OFFSET $3",
"query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags, worker_group, wm_version FROM worker_ping\n WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n ORDER BY ping_at desc LIMIT $2 OFFSET $3",
"describe": {
"columns": [
{
@@ -35,41 +35,25 @@
},
{
"ordinal": 6,
"name": "current_job_id",
"type_info": "Uuid"
},
{
"ordinal": 7,
"name": "current_job_workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 8,
"name": "custom_tags",
"type_info": "TextArray"
},
{
"ordinal": 9,
"ordinal": 7,
"name": "worker_group",
"type_info": "Varchar"
},
{
"ordinal": 10,
"ordinal": 8,
"name": "wm_version",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "occupancy_rate",
"type_info": "Float4"
}
],
"parameters": {
"Left": [
"Int4",
"Int8",
"Int8",
"Bool"
"Int8"
]
},
"nullable": [
@@ -79,13 +63,10 @@
false,
false,
false,
null,
null,
true,
false,
false,
true
false
]
},
"hash": "e00171cc3fc8f32922562d4fc4d3db56955fbc6ee9872d4b2e99157ebed131e0"
"hash": "b38044d94e2ab03167c2f6fbb553ab3c19930ed11abf51763cd3ee378229443d"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT args FROM input WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "args",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "bbd5f968d7b62a55a7ebf7b98cfd411678ce544fdb9472d8d223235fb2818aaf"
}

View File

@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH all_users as (SELECT count(*)::INT as count FROM usr WHERE disabled IS false),\n authors as (SELECT count(distinct email)::INT as count FROM usr WHERE usr.operator IS false AND disabled IS false)\n SELECT authors.count as author_count, all_users.count - authors.count as operator_count FROM all_users, authors",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "author_count",
"type_info": "Int4"
},
{
"ordinal": 1,
"name": "operator_count",
"type_info": "Int4"
}
],
"parameters": {
"Left": []
},
"nullable": [
null,
null
]
},
"hash": "c584aeec21716f7405c355d5684ff3ac1be3f70849d67b8c2d4787e809b7ea2b"
}

View File

@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO metrics (id, value) VALUES ('author_count', $1), ('operator_count', $2)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "c6b2791dd109c7bf40a40b557d9d5a140e70c49718fcb3a776b77f8f2ab901d4"
}

View File

@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE worker_ping SET ping_at = now() WHERE worker = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "c9d97800eb0ec87df8e8959b283dacb2c6cce422365ed394375641488ceb6b65"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM schedule WHERE path = $1 AND workspace_id = $2 RETURNING 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "d1ded8b38e50eb01fa5e5e122dae48ec21856a0041f4aeb244349ab7648d306f"
}

View File

@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET canceled = true, canceled_by = $1, scheduled_for = now(), suspend = 0 WHERE id = $2 RETURNING 1 as one",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "one",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Varchar",
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "d4fb94ee8198592c24e85d29078feb5220ab367e339510ee0e83bb7b5abfd184"
}

976
backend/Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,6 @@
[package]
name = "windmill"
version = "1.325.2"
version = "1.306.2"
authors.workspace = true
edition.workspace = true
@@ -24,7 +24,7 @@ members = [
]
[workspace.package]
version = "1.325.2"
version = "1.306.2"
authors = ["Ruben Fiszel <ruben@windmill.dev>"]
edition = "2021"
@@ -50,7 +50,6 @@ parquet = ["windmill-api/parquet", "windmill-common/parquet", "windmill-worker/p
prometheus = ["windmill-common/prometheus", "windmill-api/prometheus", "windmill-worker/prometheus", "windmill-queue/prometheus"]
flow_testing = ["windmill-worker/flow_testing"]
openidconnect = ["windmill-api/openidconnect"]
cloud = ["windmill-queue/cloud"]
[dependencies]
anyhow.workspace = true
@@ -109,7 +108,7 @@ windmill-parser-sql = { path = "./parsers/windmill-parser-sql" }
windmill-parser-graphql = { path = "./parsers/windmill-parser-graphql" }
windmill-api-client = { path = "./windmill-api-client" }
axum = { version = "^0.7", features = ["multipart"] }
axum = { version = "^0.7" }
headers = "^0"
hyper = { version = "^1", features = ["full"] }
tokio = { version = "^1", features = ["full", "tracing"] }
@@ -225,7 +224,7 @@ tokenizers = "0.14.1"
candle-core = "0.3.0"
candle-transformers = "0.3.0"
candle-nn = "0.3.0"
tiberius = { git = "https://github.com/prisma/tiberius", rev = "8f66a699dfa041e7b5f736c7e94f92c945453c9e", default-features = false, features = ["rustls", "tds73", "chrono", "sql-browser-tokio"]}
tiberius = { version = "0.12.2", default-features = false, features = ["rustls", "tds73", "chrono", "sql-browser-tokio"] }
pin-project = "1"
indexmap = { version = "2.2.5", features = ["serde"]}

View File

@@ -1 +1 @@
644b6f49f087790a728a7a0a82525997b1742738
66d9cbb158ab9a5869a45ba253bf57f2cbb5ecb6

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,228 +0,0 @@
DO
$do$
DECLARE
i text;
arr text[] := array['resource', 'script', 'variable', 'schedule', 'flow', 'app', 'raw_app'];
BEGIN
FOREACH i IN ARRAY arr
LOOP
EXECUTE FORMAT(
$$
DROP POLICY IF EXISTS see_folder_extra_perms_user ON %1$I;
DROP POLICY IF EXISTS see_folder_extra_perms_user_delete ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_user ON %1$I;
DROP POLICY IF EXISTS see_member ON %1$I;
DROP POLICY IF EXISTS see_own ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_user_delete ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_groups ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_groups_delete ON %1$I;
-- New policies for select, insert, update
DROP POLICY IF EXISTS see_folder_extra_perms_user_select ON %1$I;
DROP POLICY IF EXISTS see_folder_extra_perms_user_insert ON %1$I;
DROP POLICY IF EXISTS see_folder_extra_perms_user_update ON %1$I;
DROP POLICY IF EXISTS see_own ON %1$I;
DROP POLICY IF EXISTS see_member ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_user_select ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_user_insert ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_user_update ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_groups_select ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_groups_insert ON %1$I;
DROP POLICY IF EXISTS see_extra_perms_groups_update ON %1$I;
-- Folder permissions split into select, insert, and update
CREATE POLICY see_folder_extra_perms_user_select ON %1$I FOR SELECT TO windmill_user
USING (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]));
CREATE POLICY see_folder_extra_perms_user_insert ON %1$I FOR INSERT TO windmill_user
WITH CHECK (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
CREATE POLICY see_folder_extra_perms_user_update ON %1$I FOR UPDATE TO windmill_user
USING (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
CREATE POLICY see_folder_extra_perms_user_delete ON %1$I FOR UPDATE TO windmill_user
USING (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
CREATE POLICY see_own ON %1$I FOR ALL TO windmill_user
USING (SPLIT_PART(%1$I.path, '/', 1) = 'u' AND SPLIT_PART(%1$I.path, '/', 2) = current_setting('session.user'));
CREATE POLICY see_member ON %1$I FOR ALL TO windmill_user
USING (SPLIT_PART(%1$I.path, '/', 1) = 'g' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
CREATE POLICY see_extra_perms_user_select ON %1$I FOR SELECT TO windmill_user
USING (extra_perms ? CONCAT('u/', current_setting('session.user')));
CREATE POLICY see_extra_perms_user_insert ON %1$I FOR INSERT TO windmill_user
WITH CHECK ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
CREATE POLICY see_extra_perms_user_update ON %1$I FOR UPDATE TO windmill_user
USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
CREATE POLICY see_extra_perms_user_delete ON %1$I FOR DELETE TO windmill_user
USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
CREATE POLICY see_extra_perms_groups_select ON %1$I FOR SELECT TO windmill_user
USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[]);
CREATE POLICY see_extra_perms_groups_insert ON %1$I FOR INSERT TO windmill_user
WITH CHECK (exists(
SELECT key, value FROM jsonb_each_text(extra_perms)
WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
AND value::boolean));
CREATE POLICY see_extra_perms_groups_update ON %1$I FOR UPDATE TO windmill_user
USING (exists(
SELECT key, value FROM jsonb_each_text(extra_perms)
WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
AND value::boolean));
CREATE POLICY see_extra_perms_groups_delete ON %1$I FOR DELETE TO windmill_user
USING (exists(
SELECT key, value FROM jsonb_each_text(extra_perms)
WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
AND value::boolean));
$$,
i
);
END LOOP;
END
$do$;
DROP POLICY IF EXISTS see_extra_perms_user ON folder;
DROP POLICY IF EXISTS see_extra_perms_user_select ON folder;
DROP POLICY IF EXISTS see_extra_perms_user_insert ON folder;
DROP POLICY IF EXISTS see_extra_perms_user_update ON folder;
DROP POLICY IF EXISTS see_extra_perms_user_delete ON folder;
DROP POLICY IF EXISTS see_extra_perms_groups ON folder;
DROP POLICY IF EXISTS see_extra_perms_groups_select ON folder;
DROP POLICY IF EXISTS see_extra_perms_groups_insert ON folder;
DROP POLICY IF EXISTS see_extra_perms_groups_update ON folder;
DROP POLICY IF EXISTS see_extra_perms_groups_delete ON folder;
-- Existing CREATE POLICY statements updated to reflect policy splitting for 'folder' table
CREATE POLICY see_extra_perms_user_select ON folder FOR SELECT TO windmill_user
USING (extra_perms ? CONCAT('u/', current_setting('session.user')) OR CONCAT('u/', current_setting('session.user')) = ANY(owners));
CREATE POLICY see_extra_perms_user_insert ON folder FOR INSERT TO windmill_user
WITH CHECK ((CONCAT('u/', current_setting('session.user')) = ANY(owners)));
CREATE POLICY see_extra_perms_user_update ON folder FOR UPDATE TO windmill_user
USING ((CONCAT('u/', current_setting('session.user')) = ANY(owners)));
CREATE POLICY see_extra_perms_user_delete ON folder FOR DELETE TO windmill_user
USING ((CONCAT('u/', current_setting('session.user')) = ANY(owners)));
CREATE POLICY see_extra_perms_groups_select ON folder FOR SELECT TO windmill_user
USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[] OR EXISTS (
SELECT o FROM unnest(owners) AS o
WHERE o = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])));
CREATE POLICY see_extra_perms_groups_insert ON folder FOR INSERT TO windmill_user
WITH CHECK (EXISTS (
SELECT o FROM unnest(owners) AS o
WHERE o = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])));
CREATE POLICY see_extra_perms_groups_update ON folder FOR UPDATE TO windmill_user
USING (EXISTS (
SELECT o FROM unnest(owners) AS o
WHERE o = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])));
CREATE POLICY see_extra_perms_groups_delete ON folder FOR DELETE TO windmill_user
USING (EXISTS (
SELECT o FROM unnest(owners) AS o
WHERE o = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])));
-- DO
-- $do$
-- DECLARE
-- i text;
-- arr text[] := array['resource', 'script', 'variable', 'schedule', 'flow', 'app', 'raw_app'];
-- BEGIN
-- FOREACH i IN ARRAY arr
-- LOOP
-- EXECUTE FORMAT(
-- $$
-- CREATE POLICY see_folder_extra_perms_user ON %1$I FOR ALL TO windmill_user
-- USING (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]))
-- WITH CHECK (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
-- CREATE POLICY see_folder_extra_perms_user_delete ON %1$I AS RESTRICTIVE FOR DELETE TO windmill_user
-- USING (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
-- CREATE POLICY see_own ON %1$I FOR ALL TO windmill_user
-- USING (SPLIT_PART(%1$I.path, '/', 1) = 'u' AND SPLIT_PART(%1$I.path, '/', 2) = current_setting('session.user'));
-- CREATE POLICY see_member ON %1$I FOR ALL TO windmill_user
-- USING (SPLIT_PART(%1$I.path, '/', 1) = 'g' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
-- CREATE POLICY see_extra_perms_user ON %1$I FOR ALL TO windmill_user
-- USING (extra_perms ? CONCAT('u/', current_setting('session.user')))
-- WITH CHECK ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
-- CREATE POLICY see_extra_perms_user_delete ON %1$I FOR DELETE TO windmill_user
-- USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
-- CREATE POLICY see_extra_perms_groups ON %1$I FOR ALL TO windmill_user
-- USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
-- WITH CHECK (exists(
-- SELECT key, value FROM jsonb_each_text(extra_perms)
-- WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
-- AND value::boolean));
-- CREATE POLICY see_extra_perms_groups_delete ON %1$I FOR DELETE TO windmill_user
-- USING (exists(
-- SELECT key, value FROM jsonb_each_text(extra_perms)
-- WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
-- AND value::boolean));
-- $$,
-- i
-- );
-- END LOOP;
-- END
-- $do$;
-- DROP POLICY see_extra_perms_user ON folder;
-- DROP POLICY see_extra_perms_groups ON folder;
-- CREATE POLICY see_extra_perms_user ON folder FOR ALL to windmill_user
-- USING (extra_perms ? CONCAT('u/', current_setting('session.user')) or (CONCAT('u/', current_setting('session.user')) = ANY(owners)))
-- WITH CHECK ((CONCAT('u/', current_setting('session.user')) = ANY(owners)));
-- DROP POLICY IF EXISTS see_extra_perms_user_delete ON folder;
-- CREATE POLICY see_extra_perms_user_delete ON folder AS RESTRICTIVE FOR DELETE to windmill_user
-- USING ((CONCAT('u/', current_setting('session.user')) = ANY(owners)));
-- CREATE POLICY see_extra_perms_groups ON folder FOR ALL to windmill_user
-- USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[] or (exists(
-- SELECT o FROM unnest(owners) as o
-- WHERE o = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[]))))
-- WITH CHECK (exists(
-- SELECT o FROM unnest(owners) as o
-- WHERE o = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])));
-- DROP POLICY IF EXISTS see_extra_perms_groups_delete ON folder;
-- CREATE POLICY see_extra_perms_groups_delete ON folder AS RESTRICTIVE FOR DELETE to windmill_user
-- USING (exists(
-- SELECT o FROM unnest(owners) as o
-- WHERE o = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])));

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,22 +0,0 @@
DO
$do$
DECLARE
i text;
arr text[] := array['resource', 'script', 'variable', 'schedule', 'flow', 'app', 'raw_app'];
BEGIN
FOREACH i IN ARRAY arr
LOOP
EXECUTE FORMAT(
$$
DROP POLICY IF EXISTS see_folder_extra_perms_user_delete ON %1$I;
CREATE POLICY see_folder_extra_perms_user_delete ON %1$I FOR DELETE TO windmill_user
USING (SPLIT_PART(%1$I.path, '/', 1) = 'f' AND SPLIT_PART(%1$I.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
$$,
i
);
END LOOP;
END
$do$;-- Add up migration script here

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,2 +0,0 @@
-- Add up migration script here
ALTER TABLE script ADD COLUMN no_main_func BOOLEAN;

View File

@@ -1,2 +0,0 @@
-- Add down migration script here
DROP INDEX IF EXISTS labeled_jobs_on_completed_jobs;

View File

@@ -1 +0,0 @@
-- Add up migration script here

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1 +0,0 @@
-- Add up migration script here

View File

@@ -1,2 +0,0 @@
-- Add down migration script here
ALTER TABLE script DROP COLUMN codebase;

View File

@@ -1,3 +0,0 @@
-- Add up migration script here
ALTER TABLE script ADD COLUMN codebase VARCHAR(255);

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,7 +0,0 @@
-- Add up migration script here
ALTER TABLE worker_ping
ADD COLUMN current_job_id UUID,
ADD COLUMN current_job_workspace_id VARCHAR(50),
ADD COLUMN vcpus BIGINT,
ADD COLUMN memory BIGINT,
ADD COLUMN occupancy_rate REAL;

View File

@@ -1 +0,0 @@
-- Add down migration script here

View File

@@ -1,8 +0,0 @@
-- Add up migration script here
CREATE TABLE metrics (
id VARCHAR(255) NOT NULL,
value JSONB NOT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL
);
CREATE INDEX metrics_key_idx ON metrics(id);
CREATE INDEX metrics_sort_idx ON metrics(created_at DESC);

View File

@@ -141,16 +141,5 @@
"scopes": [
"com.intuit.quickbooks.accounting"
]
},
"visma": {
"auth_url": "https://connect.visma.com/connect/authorize",
"token_url": "https://connect.visma.com/connect/token",
"scopes": [
"offline_access",
"vismanet_erp_interactive_api:create",
"vismanet_erp_interactive_api:delete",
"vismanet_erp_interactive_api:read",
"vismanet_erp_interactive_api:update"
]
}
}

View File

@@ -50,9 +50,6 @@ static PYTHON_IMPORTS_REPLACEMENT: phf::Map<&'static str, &'static str> = phf_ma
"mysql" => "mysql-connector-python",
"tenable" => "pytenable",
"ns1" => "ns1-python",
"pymsql" => "PyMySQL",
"haystack" => "haystack-ai",
"github" => "PyGithub",
};
fn replace_import(x: String) -> String {
@@ -64,7 +61,7 @@ fn replace_import(x: String) -> String {
}
lazy_static! {
static ref RE: Regex = Regex::new(r"^\#\s?(\S+)\s*$").unwrap();
static ref RE: Regex = Regex::new(r"^\#\s?(\S+)$").unwrap();
}
fn process_import(module: Option<String>, path: &str, level: usize) -> Vec<String> {

View File

@@ -292,30 +292,22 @@ pub fn parse_pg_typ(typ: &str) -> Typ {
Typ::List(Box::new(base_typ))
} else {
match typ {
"varchar" | "character varying" => Typ::Str(None),
"varchar" => Typ::Str(None),
"text" => Typ::Str(None),
"int" | "integer" | "int4" => Typ::Int,
"int" => Typ::Int,
"bigint" => Typ::Int,
"bool" | "boolean" => Typ::Bool,
"char" | "character" => Typ::Str(None),
"json" | "jsonb" => Typ::Object(vec![]),
"smallint" | "int2" => Typ::Int,
"smallserial" | "serial2" => Typ::Int,
"serial" | "serial4" => Typ::Int,
"bigserial" | "serial8" => Typ::Int,
"real" | "float4" => Typ::Float,
"double" | "double precision" | "float8" => Typ::Float,
"numeric" | "decimal" => Typ::Float,
"bool" => Typ::Bool,
"char" => Typ::Str(None),
"smallint" => Typ::Int,
"smallserial" => Typ::Int,
"serial" => Typ::Int,
"bigserial" => Typ::Int,
"real" => Typ::Float,
"double precision" => Typ::Float,
"numeric" => Typ::Float,
"decimal" => Typ::Float,
"oid" => Typ::Int,
"date"
| "time"
| "timetz"
| "time with time zone"
| "time without time zone"
| "timestamp"
| "timestamptz"
| "timestamp with time zone"
| "timestamp without time zone" => Typ::Datetime,
"date" | "time" | "timestamp" | "timestamptz" => Typ::Datetime,
_ => Typ::Str(None),
}
}

View File

@@ -3,7 +3,7 @@
"collaborators": [
"Ruben Fiszel <ruben@windmill.dev>"
],
"version": "1.318.0",
"version": "1.286.2",
"files": [
"windmill_parser_wasm_bg.wasm",
"windmill_parser_wasm.js",

View File

@@ -561,6 +561,10 @@ async function __wbg_load(module, imports) {
function __wbg_get_imports() {
const imports = {};
imports.wbg = {};
imports.wbg.__wbg_eval_89aaea39f7e976e8 = function(arg0, arg1) {
const ret = eval(getStringFromWasm0(arg0, arg1));
return addHeapObject(ret);
};
imports.wbg.__wbindgen_object_drop_ref = function(arg0) {
takeObject(arg0);
};
@@ -612,10 +616,6 @@ function __wbg_get_imports() {
const ret = getObject(arg0) in getObject(arg1);
return ret;
};
imports.wbg.__wbg_eval_33c4985197d1feaf = function(arg0, arg1) {
const ret = eval(getStringFromWasm0(arg0, arg1));
return addHeapObject(ret);
};
imports.wbg.__wbindgen_jsval_loose_eq = function(arg0, arg1) {
const ret = getObject(arg0) == getObject(arg1);
return ret;

View File

@@ -18,11 +18,11 @@ use tokio::fs::DirBuilder;
use windmill_api::HTTP_CLIENT;
use windmill_common::{
global_settings::{
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, ENV_SETTINGS,
EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING,
LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CUSTOM_TAGS_SETTING,
DEFAULT_TAGS_PER_WORKSPACE_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING,
EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING,
JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING,
NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
},
@@ -47,11 +47,10 @@ use windmill_worker::{
use crate::monitor::{
initial_load, load_keep_job_dir, load_require_preexisting_user, load_tag_per_workspace_enabled,
monitor_db, monitor_pool, reload_base_url_setting, reload_bunfig_install_scopes_setting,
reload_critical_error_channels_setting, reload_extra_pip_index_url_setting,
reload_hub_base_url_setting, reload_job_default_timeout_setting, reload_license_key,
reload_npm_config_registry_setting, reload_pip_index_url_setting,
reload_retention_period_setting, reload_scim_token_setting, reload_server_config,
reload_worker_config,
reload_extra_pip_index_url_setting, reload_hub_base_url_setting,
reload_job_default_timeout_setting, reload_license_key, reload_npm_config_registry_setting,
reload_pip_index_url_setting, reload_retention_period_setting, reload_scim_token_setting,
reload_server_config, reload_worker_config,
};
#[cfg(feature = "parquet")]
@@ -504,11 +503,6 @@ Windmill Community Edition {GIT_VERSION}
tracing::error!(error = %e, "Could not reload hub base url setting");
}
},
CRITICAL_ERROR_CHANNELS_SETTING => {
if let Err(e) = reload_critical_error_channels_setting(&db).await {
tracing::error!(error = %e, "Could not reload critical error emails setting");
}
},
a @_ => {
tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a);
}

View File

@@ -8,7 +8,7 @@ use std::{
};
use rsmq_async::MultiplexedRsmq;
use serde::{de::DeserializeOwned, Serialize};
use serde::de::DeserializeOwned;
use sqlx::{Pool, Postgres};
use tokio::{
join,
@@ -22,17 +22,15 @@ use windmill_api::{
DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN,
};
use windmill_common::{
ee::CriticalErrorChannel,
error,
flow_status::FlowStatusModule,
global_settings::{
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
DEFAULT_TAGS_PER_WORKSPACE_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING,
PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING,
LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
},
jobs::QueuedJob,
oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH,
@@ -42,8 +40,7 @@ use windmill_common::{
load_worker_config, reload_custom_tags_setting, DEFAULT_TAGS_PER_WORKSPACE, SERVER_CONFIG,
WORKER_CONFIG,
},
BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL,
METRICS_DEBUG_ENABLED, METRICS_ENABLED,
BASE_URL, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
};
use windmill_queue::cancel_job;
use windmill_worker::{
@@ -146,10 +143,6 @@ pub async fn initial_load(
tracing::error!("Error reloading hub base url: {:?}", e)
}
if let Err(e) = reload_critical_error_channels_setting(&db).await {
tracing::error!("Could not reload critical error emails setting: {:?}", e);
}
#[cfg(feature = "parquet")]
if !_is_agent {
reload_s3_cache_setting(&db).await;
@@ -678,182 +671,39 @@ pub async fn monitor_db(
};
let expose_queue_metrics_f = async {
if !initial_load && server_mode {
if !initial_load
&& METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed)
&& server_mode
{
expose_queue_metrics(&db).await;
}
};
let save_usage_metrics_f = async {
if !initial_load && server_mode {
save_usage_metrics(&db).await;
}
};
join!(
expired_items_f,
zombie_jobs_f,
expose_queue_metrics_f,
save_usage_metrics_f,
verify_license_key_f
);
}
pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
let tx = db.begin().await;
if let Ok(mut tx) = tx {
let last_check = sqlx::query_scalar!(
"SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1"
)
.fetch_optional(db)
.await
.unwrap_or(Some(chrono::Utc::now()));
let metrics_enabled = METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed);
let save_metrics = last_check
.map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::seconds(25))
.unwrap_or(true);
if metrics_enabled || save_metrics {
let queue_counts = sqlx::query!(
"SELECT tag, count(*) as count FROM queue WHERE
scheduled_for <= now() - ('3 seconds')::interval AND running = false
GROUP BY tag"
)
.fetch_all(&mut *tx)
.await
.ok()
.unwrap_or_else(|| vec![]);
for q in queue_counts {
let count = q.count.unwrap_or(0);
let tag = q.tag;
if metrics_enabled {
let metric = (*QUEUE_COUNT).with_label_values(&[&tag]);
metric.set(count as i64);
}
// save queue_count and delay metrics per tag
if save_metrics {
sqlx::query!(
"INSERT INTO metrics (id, value) VALUES ($1, $2)",
format!("queue_count_{}", tag),
serde_json::json!(count)
)
.execute(&mut *tx)
.await
.ok();
if count > 0 {
sqlx::query!(
"INSERT INTO metrics (id, value)
VALUES ($1, to_jsonb((SELECT EXTRACT(EPOCH FROM now() - scheduled_for)
FROM queue WHERE tag = $2 AND running = false AND scheduled_for <= now() - ('3 seconds')::interval
ORDER BY priority DESC NULLS LAST, scheduled_for, created_at LIMIT 1)))",
format!("queue_delay_{}", tag),
tag
).execute(&mut *tx).await.ok();
}
}
}
}
// clean queue metrics older than 14 days
sqlx::query!(
"DELETE FROM metrics WHERE id LIKE 'queue_%' AND created_at < NOW() - INTERVAL '14 day'"
).execute(&mut *tx).await.ok();
tx.commit().await.ok();
let queue_counts = sqlx::query!(
"SELECT tag, count(*) as count FROM queue WHERE
scheduled_for <= now() - ('3 seconds')::interval AND running = false
GROUP BY tag"
)
.fetch_all(db)
.await
.ok()
.unwrap_or_else(|| vec![]);
for q in queue_counts {
let count = q.count.unwrap_or(0);
let tag = q.tag;
let metric = (*QUEUE_COUNT).with_label_values(&[&tag]);
metric.set(count as i64);
}
}
#[derive(Serialize)]
struct WorkerUsage {
worker: String,
worker_instance: String,
vcpus: Option<i64>,
memory: Option<i64>,
}
pub async fn save_usage_metrics(db: &Pool<Postgres>) {
let tx = db.begin().await;
if let Ok(mut tx) = tx {
let last_check = sqlx::query_scalar!(
"SELECT created_at FROM metrics WHERE id = 'author_count' ORDER BY created_at DESC LIMIT 1"
)
.fetch_optional(db)
.await
.unwrap_or(Some(chrono::Utc::now()));
let random_nb = rand::random::<i64>();
// save author and operator count every ~24 hours
if last_check
.map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::hours(24) - chrono::Duration::minutes(random_nb % 60))
.unwrap_or(true)
{
let counts = sqlx::query!(
"WITH all_users as (SELECT count(*)::INT as count FROM usr WHERE disabled IS false),
authors as (SELECT count(distinct email)::INT as count FROM usr WHERE usr.operator IS false AND disabled IS false)
SELECT authors.count as author_count, all_users.count - authors.count as operator_count FROM all_users, authors"
)
.fetch_one(&mut *tx)
.await
.ok();
if let Some(counts) = counts {
sqlx::query!(
"INSERT INTO metrics (id, value) VALUES ('author_count', $1), ('operator_count', $2)",
serde_json::json!(counts.author_count),
serde_json::json!(counts.operator_count)
)
.execute(&mut *tx)
.await
.ok();
}
// clean metrics older than 6 months (including worker usage)
sqlx::query!(
"DELETE FROM metrics
WHERE (id = 'author_count' OR id = 'operator_count' OR id = 'worker_usage') AND created_at < NOW() - INTERVAL '6 month'"
)
.execute(&mut *tx)
.await
.ok();
}
// save worker usage every ~60 minutes
if last_check
.map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::minutes(60) - chrono::Duration::seconds(random_nb % 300))
.unwrap_or(true)
{
let worker_usage = sqlx::query_as!(
WorkerUsage,
"SELECT worker, worker_instance, vcpus, memory FROM worker_ping WHERE ping_at > NOW() - INTERVAL '2 minutes'"
)
.fetch_all(&mut *tx)
.await
.ok();
if let Some(worker_usage) = worker_usage {
sqlx::query!(
"INSERT INTO metrics (id, value) VALUES ('worker_usage', $1)",
serde_json::json!(worker_usage)
)
.execute(&mut *tx)
.await
.ok();
}
}
tx.commit().await.ok();
}
}
pub async fn reload_server_config(db: &Pool<Postgres>) {
let config = load_server_config(&db).await;
if let Err(e) = config {
@@ -1224,27 +1074,3 @@ pub async fn reload_hub_base_url_setting(db: &DB, server_mode: bool) -> error::R
Ok(())
}
pub async fn reload_critical_error_channels_setting(db: &DB) -> error::Result<()> {
let critical_error_channels =
load_value_from_global_settings(db, CRITICAL_ERROR_CHANNELS_SETTING).await?;
let critical_error_channels = if let Some(q) = critical_error_channels {
if let Ok(v) = serde_json::from_value::<Vec<CriticalErrorChannel>>(q.clone()) {
v
} else {
tracing::error!(
"Could not parse critical_error_emails setting as an array of channels, found: {:#?}",
&q
);
vec![]
}
} else {
vec![]
};
let mut l = CRITICAL_ERROR_CHANNELS.write().await;
*l = critical_error_channels;
Ok(())
}

View File

@@ -1026,7 +1026,7 @@ async fn listen_for_uuid_on(
async fn completed_job(uuid: Uuid, db: &Pool<Postgres>) -> CompletedJob {
sqlx::query_as::<_, CompletedJob>("SELECT *, result->'wm_labels' as labels FROM completed_job WHERE id = $1").bind(uuid)
sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1").bind(uuid)
.fetch_one(db)
.await
.unwrap()
@@ -1663,7 +1663,6 @@ func main(derp string) (string, error) {
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
@@ -1695,7 +1694,6 @@ echo "hello $msg"
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
@@ -1724,7 +1722,6 @@ def main():
.to_owned();
let job = JobPayload::Code(RawCode {
hash: None,
content,
path: None,
language: ScriptLang::Python3,
@@ -1759,7 +1756,6 @@ def main():
.to_owned();
let job = JobPayload::Code(RawCode {
hash: None,
content,
path: None,
language: ScriptLang::Python3,
@@ -1793,7 +1789,6 @@ def main():
.to_owned();
let job = JobPayload::Code(RawCode {
hash: None,
content,
path: None,
language: ScriptLang::Python3,
@@ -3169,8 +3164,6 @@ async fn run_deployed_relative_imports(db: &Pool<Postgres>, script_content: Stri
deployment_message: None,
concurrency_key: None,
visible_to_runner_only: None,
no_main_func: None,
codebase: None
},
).await.unwrap();
@@ -3218,7 +3211,6 @@ async fn run_preview_relative_imports(db: &Pool<Postgres>, script_content: Strin
let db2 = db.clone();
in_test_worker(&db, async move {
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content: script_content,
path: Some("f/system/test_import".to_string()),
language,

View File

@@ -72,6 +72,7 @@ async_zip.workspace = true
rsmq_async.workspace = true
regex.workspace = true
bytes.workspace = true
mail-send.workspace = true
samael = { workspace = true, optional = true }
async-recursion.workspace = true
rsa.workspace = true

View File

@@ -1,7 +1,7 @@
openapi: "3.0.3"
info:
version: 1.325.2
version: 1.306.2
title: Windmill API
contact:
@@ -3401,13 +3401,6 @@ paths:
in: query
schema:
type: boolean
- name: hide_without_main
description: |
(default false)
hide the scripts without an exported main function
in: query
schema:
type: boolean
- name: is_template
description: |
(default regardless)
@@ -4907,8 +4900,6 @@ paths:
- language
force_viewer_static_fields:
type: object
force_viewer_one_of_fields:
type: object
required:
- args
- component
@@ -5250,11 +5241,6 @@ paths:
in: query
schema:
type: boolean
- name: is_not_schedule
description: is not a scheduled job
in: query
schema:
type: boolean
responses:
"200":
description: All queued jobs
@@ -5340,7 +5326,6 @@ paths:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/OrderDesc"
- $ref: "#/components/parameters/CreatedBy"
- $ref: "#/components/parameters/Label"
- $ref: "#/components/parameters/ParentJob"
- $ref: "#/components/parameters/ScriptExactPath"
- $ref: "#/components/parameters/ScriptStartPath"
@@ -5370,11 +5355,6 @@ paths:
in: query
schema:
type: boolean
- name: is_not_schedule
description: is not a scheduled job
in: query
schema:
type: boolean
responses:
"200":
description: All completed jobs
@@ -5394,7 +5374,6 @@ paths:
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/CreatedBy"
- $ref: "#/components/parameters/Label"
- $ref: "#/components/parameters/ParentJob"
- $ref: "#/components/parameters/ScriptExactPath"
- $ref: "#/components/parameters/ScriptStartPath"
@@ -5437,11 +5416,6 @@ paths:
in: query
schema:
type: boolean
- name: is_not_schedule
description: is not a scheduled job
in: query
schema:
type: boolean
responses:
"200":
description: All jobs
@@ -5560,26 +5534,6 @@ paths:
flow_status:
$ref: "#/components/schemas/WorkflowStatusRecord"
/w/{workspace}/jobs_u/get_log_file/{path}:
get:
summary: get log file from object store
operationId: getLogFileFromStore
tags:
- job
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- name: path
in: path
required: true
schema:
type: string
responses:
"200":
description: job log
content:
text/plain:
type: string
/w/{workspace}/jobs_u/get_flow_debug_info/{id}:
get:
summary: get flow debug info
@@ -6959,40 +6913,6 @@ paths:
schema:
type: boolean
/workers/queue_metrics:
get:
summary: get queue metrics
operationId: getQueueMetrics
tags:
- worker
responses:
"200":
description: metrics
content:
application/json:
schema:
type: array
items:
type: object
properties:
id:
type: string
values:
type: array
items:
type: object
properties:
created_at:
type: string
value:
type: number
required:
- created_at
- value
required:
- id
- values
/configs/list_worker_groups:
get:
summary: list worker groups
@@ -7305,26 +7225,6 @@ paths:
items:
$ref: "#/components/schemas/Input"
/w/{workspace}/inputs/{jobOrInputId}/args:
get:
summary: Get args from history or saved input
operationId: getArgsFromHistoryOrSavedInput
tags:
- input
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- name: jobOrInputId
in: path
required: true
schema:
type: string
responses:
"200":
description: args
content:
application/json:
schema: {}
/w/{workspace}/inputs/list:
get:
summary: List saved Inputs for a Runnable
@@ -8105,12 +8005,6 @@ components:
in: query
schema:
type: string
Label:
name: label
description: mask to filter exact matching job's label (job labels are completed jobs with as a result an object containing a string in the array at key 'wm_labels')
in: query
schema:
type: string
ParentJob:
name: parent_job
description:
@@ -8442,11 +8336,6 @@ components:
type: boolean
visible_to_runner_only:
type: boolean
no_main_func:
type: boolean
codebase:
type: string
required:
- hash
- path
@@ -8462,8 +8351,6 @@ components:
- language
- kind
- starred
- no_main_func
NewScript:
type: object
@@ -8537,10 +8424,6 @@ components:
type: string
visible_to_runner_only:
type: boolean
no_main_func:
type: boolean
codebase:
type: string
required:
- path
- summary
@@ -8842,10 +8725,6 @@ components:
type: string
priority:
type: integer
labels:
type: array
items:
type: string
required:
- id
- created_by
@@ -8863,21 +8742,15 @@ components:
- tag
Job:
oneOf:
- allOf:
allOf:
- oneOf:
- $ref: "#/components/schemas/CompletedJob"
- type: object
properties:
type:
type: string
enum: [CompletedJob]
- allOf:
- $ref: "#/components/schemas/QueuedJob"
- type: object
properties:
type:
type: string
enum: [QueuedJob]
- type: object
properties:
type:
type: string
enum: [CompletedJob, QueuedJob]
discriminator:
propertyName: type
@@ -9738,12 +9611,6 @@ components:
type: string
wm_version:
type: string
current_job_id:
type: string
current_job_workspace_id:
type: string
occupancy_rate:
type: number
required:
- worker
- worker_instance
@@ -9853,11 +9720,6 @@ components:
- $ref: "../../openflow.openapi.yaml#/components/schemas/OpenFlow"
- $ref: "#/components/schemas/FlowMetadata"
ExtraPerms:
type: object
additionalProperties:
type: boolean
FlowMetadata:
type: object
properties:
@@ -9873,7 +9735,9 @@ components:
archived:
type: boolean
extra_perms:
$ref: "#/components/schemas/ExtraPerms"
type: object
additionalProperties:
type: boolean
starred:
type: boolean
draft_only:
@@ -9956,10 +9820,6 @@ components:
type: object
additionalProperties:
type: object
triggerables_v2:
type: object
additionalProperties:
type: object
execution_mode:
type: string
enum: [viewer, publisher, anonymous]
@@ -10051,8 +9911,7 @@ components:
created_at:
type: string
format: date-time
value:
type: object
value: {}
policy:
$ref: "#/components/schemas/Policy"
execution_mode:

View File

@@ -143,7 +143,6 @@ pub struct AppHistoryUpdate {
}
pub type StaticFields = HashMap<String, Box<RawValue>>;
pub type OneOfFields = HashMap<String, Vec<Box<RawValue>>>;
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
#[serde(rename_all = "lowercase")]
@@ -153,12 +152,6 @@ pub enum ExecutionMode {
Viewer,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct PolicyTriggerableInputs {
static_inputs: StaticFields,
one_of_inputs: OneOfFields,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Policy {
pub on_behalf_of: Option<String>,
@@ -167,10 +160,7 @@ pub struct Policy {
// - script/<path>
// - flow/<path>
// - rawscript/<sha256>
#[serde(skip_serializing_if = "Option::is_none")]
pub triggerables: Option<HashMap<String, StaticFields>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub triggerables_v2: Option<HashMap<String, PolicyTriggerableInputs>>,
pub triggerables: HashMap<String, StaticFields>,
pub execution_mode: ExecutionMode,
}
@@ -944,7 +934,6 @@ pub struct ExecuteApp {
pub raw_code: Option<RawCode>,
// if set, the app is executed as viewer with the given static fields
pub force_viewer_static_fields: Option<StaticFields>,
pub force_viewer_one_of_fields: Option<OneOfFields>,
}
fn digest(code: &str) -> String {
@@ -977,56 +966,39 @@ async fn execute_component(
let path = path.to_path();
let policy = match payload.clone() {
ExecuteApp {
force_viewer_static_fields: Some(static_fields),
force_viewer_one_of_fields: Some(one_of_fields),
..
} => {
let mut hm = HashMap::new();
let policy = if let Some(static_fields) = payload.clone().force_viewer_static_fields {
let mut hm = HashMap::new();
if let Some(path) = payload.path.clone() {
hm.insert(
format!("{}:{path}", payload.component),
PolicyTriggerableInputs {
static_inputs: static_fields,
one_of_inputs: one_of_fields,
},
);
} else {
hm.insert(
format!(
"{}:{}",
payload.component,
digest(payload.raw_code.clone().unwrap().content.as_str())
),
PolicyTriggerableInputs {
static_inputs: static_fields,
one_of_inputs: one_of_fields,
},
);
}
Policy {
execution_mode: ExecutionMode::Viewer,
triggerables: None,
triggerables_v2: Some(hm),
on_behalf_of: None,
on_behalf_of_email: None,
}
if let Some(path) = payload.path.clone() {
hm.insert(format!("{}:{path}", payload.component), static_fields);
} else {
hm.insert(
format!(
"{}:{}",
payload.component,
digest(payload.raw_code.clone().unwrap().content.as_str())
),
static_fields,
);
}
_ => {
let policy_o = sqlx::query_scalar!(
"SELECT policy from app WHERE path = $1 AND workspace_id = $2",
path,
&w_id
)
.fetch_optional(&db)
.await?;
let policy = not_found_if_none(policy_o, "App", path)?;
serde_json::from_value::<Policy>(policy).map_err(to_anyhow)?
Policy {
execution_mode: ExecutionMode::Viewer,
triggerables: hm,
on_behalf_of: None,
on_behalf_of_email: None,
}
} else {
let policy_o = sqlx::query_scalar!(
"SELECT policy from app WHERE path = $1 AND workspace_id = $2",
path,
&w_id
)
.fetch_optional(&db)
.await?;
let policy = not_found_if_none(policy_o, "App", path)?;
serde_json::from_value::<Policy>(policy).map_err(to_anyhow)?
};
let (username, permissioned_as, email) = match policy.execution_mode {
@@ -1161,106 +1133,22 @@ fn build_args(
path: String,
args: HashMap<String, Box<RawValue>>,
) -> Result<PushArgs<HashMap<String, Box<RawValue>>>> {
let key = format!("{}:{}", component, &path);
let (static_inputs, one_of_inputs) = match policy {
Policy { triggerables_v2: Some(t), .. } => {
let PolicyTriggerableInputs { static_inputs, one_of_inputs } = t
.get(&key)
.or_else(|| t.get(&path))
.map(|x| x.clone())
.or_else(|| {
if matches!(policy.execution_mode, ExecutionMode::Viewer) {
Some(PolicyTriggerableInputs {
static_inputs: HashMap::new(),
one_of_inputs: HashMap::new(),
})
} else {
None
}
})
.ok_or_else(|| {
Error::BadRequest(format!("path {} is not allowed in the app policy", path))
})?;
(static_inputs, one_of_inputs)
}
Policy { triggerables: Some(t), .. } => {
let static_inputs = t
.get(&key)
.or_else(|| t.get(&path))
.map(|x| x.clone())
.or_else(|| {
if matches!(policy.execution_mode, ExecutionMode::Viewer) {
Some(HashMap::new())
} else {
None
}
})
.ok_or_else(|| {
Error::BadRequest(format!("path {} is not allowed in the app policy", path))
})?;
(static_inputs, HashMap::new())
}
_ => Err(Error::BadRequest(format!(
"Policy is missing triggerables for {}",
key
)))?,
};
let mut args = args.clone();
let mut safe_args = HashMap::<String, Box<RawValue>>::new();
for (k, v) in one_of_inputs {
if let Some(arg_val) = args.get(&k) {
let arg_str = arg_val.get();
let options_str_vec = v.iter().map(|x| x.get()).collect::<Vec<&str>>();
if options_str_vec.contains(&arg_str) {
safe_args.insert(k.to_string(), arg_val.clone());
args.remove(&k);
continue;
}
// check if multiselect
if let Ok(args_str_vec) = serde_json::from_str::<Vec<Box<RawValue>>>(arg_val.get()) {
if args_str_vec
.iter()
.all(|x| options_str_vec.contains(&x.get()))
{
safe_args.insert(k.to_string(), arg_val.clone());
args.remove(&k);
continue;
}
}
return Err(Error::BadRequest(format!(
"argument {} with value {} must be one of [{}]",
k,
arg_str,
options_str_vec.join(",")
)));
}
}
// disallow var and res access in args coming from the user for security reasons
let mut safe_args: HashMap<String, Box<RawValue>> = args.clone();
for (k, v) in args {
let arg_str = serde_json::to_string(&v).unwrap_or_else(|_| "".to_string());
if !arg_str.contains("$var:") && !arg_str.contains("$res:") {
safe_args.insert(k.to_string(), v);
} else {
let args_str = serde_json::to_string(&v).unwrap_or_else(|_| "".to_string());
if args_str.contains("$var:") || args_str.contains("$res:") {
safe_args.insert(
k.to_string(),
RawValue::from_string(
arg_str
args_str
.replace(
"$var:",
"The following variable has been omitted for security reasons: ",
"The following variable has been ommited for security reasons: ",
)
.replace(
"$res:",
"The following resource has been omitted for security reasons: ",
"The following resource has been ommited for security reasons: ",
),
)
.map_err(|e| {
@@ -1272,8 +1160,24 @@ fn build_args(
);
}
}
let key = format!("{}:{}", component, &path);
let static_args = policy
.triggerables
.get(&key)
.or_else(|| policy.triggerables.get(&path))
.map(|x| x.clone())
.or_else(|| {
if matches!(policy.execution_mode, ExecutionMode::Viewer) {
Some(HashMap::new())
} else {
None
}
})
.ok_or_else(|| {
Error::BadRequest(format!("path {} is not allowed in the app policy", path))
})?;
let mut extra = HashMap::new();
for (k, v) in static_inputs {
for (k, v) in static_args {
extra.insert(k.to_string(), v.to_owned());
}
Ok(PushArgs { extra, args: sqlx::types::Json(safe_args) })

View File

@@ -136,15 +136,6 @@ impl Migrate for CustomMigrator {
migration.version,
migration.description
);
if migration.version == 20240424083501 {
tracing::info!("Special migration to add index concurrently on job labels 2");
sqlx::query!(
"DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs"
).execute(&mut *self.inner).await?;
sqlx::query!(
"CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON completed_job USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_label';"
).execute(&mut *self.inner).await?;
}
let r = self.inner.apply(migration).await;
tracing::info!("Finished applying migration {}", migration.version);
r

View File

@@ -543,7 +543,7 @@ async fn update_flow(
clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?;
if schedule.enabled {
tx = push_scheduled_job(&db, tx, &schedule).await?;
tx = push_scheduled_job(&db, tx, schedule).await?;
}
}
@@ -955,7 +955,7 @@ mod tests {
continue_on_error: None,
},
],
failure_module: Some(Box::new(FlowModule {
failure_module: Some(FlowModule {
id: "d".to_string(),
value: FlowModuleValue::Script {
path: "test".to_string(),
@@ -977,7 +977,7 @@ mod tests {
priority: None,
delete_after_use: None,
continue_on_error: None,
})),
}),
same_worker: false,
concurrent_limit: None,
concurrency_time_window_s: None,

View File

@@ -271,7 +271,7 @@ async fn update_folder(
Extension(user_db): Extension<UserDB>,
Extension(webhook): Extension<WebhookShared>,
Path((w_id, name)): Path<(String, String)>,
Json(mut ng): Json<UpdateFolder>,
Json(ng): Json<UpdateFolder>,
) -> Result<String> {
use sql_builder::prelude::*;
@@ -282,25 +282,6 @@ async fn update_folder(
if let Some(display_name) = ng.display_name {
sqlb.set("display_name", "?".bind(&display_name));
}
if !authed.is_admin {
let prefixed_username = format!("u/{}", authed.username);
if ng.owners.as_ref().is_some_and(|x| {
!x.contains(&prefixed_username)
&& !authed.groups.iter().any(|g| x.contains(&format!("g/{g}")))
}) {
ng.owners.as_mut().unwrap().push(prefixed_username.clone());
if ng.extra_perms.is_none() {
ng.extra_perms = Some(serde_json::Value::Object(serde_json::Map::new()));
}
ng.extra_perms
.as_mut()
.unwrap()
.as_object_mut()
.unwrap()
.insert(prefixed_username, serde_json::json!(true));
}
}
if let Some(owners) = ng.owners {
sqlb.set(
"owners",
@@ -329,16 +310,9 @@ async fn update_folder(
.sql()
.map_err(|e| error::Error::InternalErr(e.to_string()))?;
let nfolder = sqlx::query_as::<_, Folder>(&sql)
.fetch_optional(&mut *tx)
.fetch_one(&mut *tx)
.await?;
let nfolder = nfolder.ok_or_else(|| {
windmill_common::error::Error::NotAuthorized(format!(
"You are not an owner of {} and hence cannot modify it",
name
))
})?;
if let Some(extra_perms) = nfolder.extra_perms.as_object() {
for o in nfolder.owners {
if !extra_perms
@@ -504,22 +478,13 @@ async fn delete_folder(
not_found_if_none(get_folderopt(&mut tx, &w_id, &name).await?, "Folder", &name)?;
let del = sqlx::query_scalar!(
"DELETE FROM folder WHERE name = $1 AND workspace_id = $2 RETURNING 1",
sqlx::query!(
"DELETE FROM folder WHERE name = $1 AND workspace_id = $2",
name,
w_id
)
.fetch_optional(&mut *tx)
.await?
.flatten();
if del.is_none() {
return Err(windmill_common::error::Error::NotAuthorized(format!(
"Not authorized to delete folder {}",
name
)));
}
.execute(&mut *tx)
.await?;
audit_log(
&mut *tx,
&authed.username,

View File

@@ -25,7 +25,7 @@ use windmill_common::{
error::JsonResult,
jobs::JobKind,
scripts::to_i64,
utils::{not_found_if_none, paginate, Pagination},
utils::{paginate, Pagination},
};
pub fn workspaced_service() -> Router {
Router::new()
@@ -34,10 +34,6 @@ pub fn workspaced_service() -> Router {
.route("/create", post(create_input))
.route("/update", post(update_input))
.route("/delete/:id", post(delete_input))
.route(
"/:job_or_input_id/args",
get(get_args_from_history_or_saved_input),
)
}
#[derive(Debug, sqlx::FromRow, Serialize, Deserialize)]
@@ -127,7 +123,7 @@ async fn get_input_history(
let mut tx = user_db.begin(&authed).await?;
let sql = &format!(
"select id, created_at, created_by, CASE WHEN args is null or pg_column_size(args) < 40000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, success from completed_job \
"select id, created_at, created_by, args, success from completed_job \
where {} = $1 and job_kind = $2 and workspace_id = $3 \
order by created_at desc limit $4 offset $5",
r.runnable_type.column_name()
@@ -171,27 +167,6 @@ async fn get_input_history(
Ok(Json(inputs))
}
async fn get_args_from_history_or_saved_input(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, job_or_input_id)): Path<(String, Uuid)>,
) -> JsonResult<Option<Value>> {
let mut tx = user_db.begin(&authed).await?;
let result_o = sqlx::query_scalar!(
"SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT args FROM input WHERE id = $1 AND workspace_id = $2",
job_or_input_id,
w_id
)
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
let result = not_found_if_none(result_o, "Input args", job_or_input_id.to_string())?;
Ok(Json(result))
}
async fn list_saved_inputs(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -204,7 +179,7 @@ async fn list_saved_inputs(
let mut tx = user_db.begin(&authed).await?;
let rows = sqlx::query_as::<_, InputRow>(
"select id, workspace_id, runnable_id, runnable_type, name, CASE WHEN pg_column_size(args) < 40000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, created_at, created_by, is_public from input \
"select * from input \
where runnable_id = $1 and runnable_type = $2 and workspace_id = $3 \
and (is_public IS true OR created_by = $4) \
order by created_at desc limit $5 offset $6",

View File

@@ -19,9 +19,6 @@ use windmill_common::jobs::{
format_completed_job_result, format_result, CompletedJobWithFormattedResult, FormattedResult,
ENTRYPOINT_OVERRIDE,
};
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use windmill_common::scripts::PREVIEW_IS_CODEBASE_HASH;
use windmill_common::variables::get_workspace_key;
use crate::db::ApiAuthed;
@@ -166,7 +163,6 @@ pub fn workspaced_service() -> Router {
.layer(cors.clone()),
)
.route("/run/preview", post(run_preview_script))
.route("/run/preview_bundle", post(run_bundle_preview_script))
.route("/add_batch_jobs/:n", post(add_batch_jobs))
.route("/run/preview_flow", post(run_preview_flow_job))
.route(
@@ -255,7 +251,6 @@ pub fn global_service() -> Router {
get(get_completed_job_result_maybe),
)
.route("/getupdate/:id", get(get_job_update))
.route("/get_log_file/*file_path", get(get_log_file))
.route("/queue/cancel/:id", post(cancel_job_api))
.route(
"/queue/cancel_persistent/*script_path",
@@ -589,7 +584,6 @@ fn generate_get_job_query(no_logs: bool, table: &str) -> String {
result,
deleted,
is_skipped,
result->'wm_labels' as labels,
CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result"
} else {
"scheduled_for,
@@ -784,8 +778,6 @@ pub struct ListableCompletedJob {
pub tag: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
#[serde(skip_serializing_if = "Option::is_none")]
pub labels: Option<serde_json::Value>,
}
#[derive(Deserialize, Clone)]
@@ -846,7 +838,6 @@ pub struct ListQueueQuery {
pub all_workspaces: Option<bool>,
pub is_flow_step: Option<bool>,
pub has_null_parent: Option<bool>,
pub is_not_schedule: Option<bool>,
}
fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> SqlBuilder {
@@ -951,10 +942,6 @@ fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> Sq
sqlb.and_where_le("scheduled_for", "now()");
}
if lq.is_not_schedule.unwrap_or(false) {
sqlb.and_where("schedule_path IS null");
}
sqlb
}
@@ -1025,62 +1012,49 @@ async fn cancel_all(
) -> error::JsonResult<Vec<Uuid>> {
require_admin(authed.is_admin, &authed.username)?;
let jobs = sqlx::query!(
"SELECT id, running, is_flow_step FROM queue WHERE scheduled_for < now() AND workspace_id = $1 AND schedule_path IS NULL",
let mut jobs = sqlx::query!(
"UPDATE queue SET canceled = true, canceled_by = $2, scheduled_for = now(), suspend = 0 WHERE scheduled_for < now() AND workspace_id = $1 AND schedule_path IS NULL RETURNING id, running, is_flow_step",
w_id,
authed.username
)
.fetch_all(&db)
.await?;
let username = authed.username;
let mut uuids = vec![];
for j in jobs.iter() {
let r = sqlx::query!(
"UPDATE queue SET canceled = true, canceled_by = $1, scheduled_for = now(), suspend = 0 WHERE id = $2 RETURNING 1 as one",
username,
j.id,
)
.fetch_optional(&db)
.await;
if !j.running && !j.is_flow_step.unwrap_or(false) {
let e = serde_json::json!({"message": format!("Job canceled: cancel_all by {username}"), "name": "Canceled", "reason": "cancel_all", "canceler": username});
let job_running = get_queued_job(&j.id, &w_id, &db).await?;
if r.as_ref().is_ok_and(|x| x.is_some()) {
uuids.push(j.id);
if !j.running && !j.is_flow_step.unwrap_or(false) {
let e = serde_json::json!({"message": format!("Job canceled: cancel_all by {username}"), "name": "Canceled", "reason": "cancel_all", "canceler": username});
let job_running = get_queued_job(&j.id, &w_id, &db).await?;
if let Some(job_running) = job_running {
append_logs(
j.id,
w_id.clone(),
format!("canceled by {username}: cancel_all"),
db.clone(),
)
.await;
let add_job = add_completed_job_error(
&db,
&job_running,
job_running.mem_peak.unwrap_or(0),
Some(CanceledBy {
username: Some(username.to_string()),
reason: Some("cancel_all".to_string()),
}),
e,
rsmq.clone(),
"server",
true,
)
.await;
if let Err(e) = add_job {
tracing::error!("Failed to add canceled job: {}", e);
}
if let Some(job_running) = job_running {
append_logs(
j.id,
w_id.clone(),
format!("canceled by {username}: cancel_all"),
db.clone(),
)
.await;
let add_job = add_completed_job_error(
&db,
&job_running,
job_running.mem_peak.unwrap_or(0),
Some(CanceledBy {
username: Some(username.to_string()),
reason: Some("cancel_all".to_string()),
}),
e,
rsmq.clone(),
"server",
true,
)
.await;
if let Err(e) = add_job {
tracing::error!("Failed to add canceled job: {}", e);
}
}
} else {
tracing::error!("Failed to cancel job: {:?} {:?}", j.id, r.err());
}
}
let uuids = jobs.iter_mut().map(|j| j.id).collect::<Vec<_>>();
Ok(Json(uuids))
}
@@ -1183,14 +1157,13 @@ async fn list_jobs(
"null as concurrent_limit",
"null as concurrency_time_window_s",
"priority",
"result->'wm_labels' as labels",
],
))
} else {
None
};
let sql = if lq.success.is_none() && lq.label.is_none() {
let sql = if lq.success.is_none() {
let sqlq = list_queue_jobs_query(
&w_id,
&ListQueueQuery {
@@ -1216,7 +1189,6 @@ async fn list_jobs(
all_workspaces: lq.all_workspaces,
is_flow_step: lq.is_flow_step,
has_null_parent: lq.has_null_parent,
is_not_schedule: lq.is_not_schedule,
},
&[
"'QueuedJob' as typ",
@@ -1250,7 +1222,6 @@ async fn list_jobs(
"concurrent_limit",
"concurrency_time_window_s",
"priority",
"null as labels",
],
);
@@ -1983,7 +1954,6 @@ struct UnifiedJob {
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
priority: Option<i16>,
labels: Option<serde_json::Value>,
}
impl<'a> From<UnifiedJob> for Job {
@@ -2021,7 +1991,6 @@ impl<'a> From<UnifiedJob> for Job {
mem_peak: uj.mem_peak,
tag: uj.tag,
priority: uj.priority,
labels: uj.labels,
}),
"QueuedJob" => Job::QueuedJob(QueuedJob {
workspace_id: uj.workspace_id,
@@ -2081,9 +2050,7 @@ enum PreviewKind {
Identity,
Http,
Noop,
Bundle,
}
#[derive(Deserialize)]
struct Preview {
content: Option<String>,
@@ -2304,7 +2271,7 @@ pub async fn restart_flow(
check_license_key_valid().await?;
let completed_job = sqlx::query_as::<_, CompletedJob>(
"SELECT *, result->'wm_labels' as labels from completed_job WHERE id = $1 and workspace_id = $2",
"SELECT * from completed_job WHERE id = $1 and workspace_id = $2",
)
.bind(job_id)
.bind(&w_id)
@@ -2426,7 +2393,6 @@ pub async fn run_workflow_as_code(
let (job_payload, tag, _delete_after_use, timeout) = match job.job_kind {
JobKind::Preview => (
JobPayload::Code(RawCode {
hash: None,
content: job.raw_code.unwrap_or_default(),
path: job.script_path,
language: job.language.unwrap_or_else(|| ScriptLang::Deno),
@@ -2528,7 +2494,7 @@ async fn run_wait_result(
w_id: String,
node_id_for_empty_return: Option<String>,
) -> error::Result<Response> {
let mut result = None;
let mut result;
let timeout = SERVER_CONFIG.read().await.timeout_wait_result.clone();
let timeout_ms = if timeout <= 0 {
2000
@@ -2552,9 +2518,7 @@ async fn run_wait_result(
)
.await
.ok();
}
if result.is_none() {
} else {
let row = sqlx::query(
"SELECT result, language, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
)
@@ -2572,6 +2536,8 @@ async fn run_wait_result(
FormattedResult::RawValue(rv) => rv,
FormattedResult::Vec(v) => Some(to_raw_value(&v)),
};
} else {
result = None;
}
}
@@ -3090,7 +3056,6 @@ async fn run_preview_script(
Some(PreviewKind::Identity) => JobPayload::Identity,
Some(PreviewKind::Noop) => JobPayload::Noop,
_ => JobPayload::Code(RawCode {
hash: None,
content: preview.content.unwrap_or_default(),
path: preview.path,
language: preview.language.unwrap_or(ScriptLang::Deno),
@@ -3125,136 +3090,6 @@ async fn run_preview_script(
Ok((StatusCode::CREATED, uuid.to_string()))
}
#[cfg(all(feature = "enterprise", feature = "parquet"))]
async fn run_bundle_preview_script(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
Path(w_id): Path<String>,
Query(run_query): Query<RunJobQuery>,
mut multipart: axum::extract::Multipart,
) -> error::Result<(StatusCode, String)> {
check_license_key_valid().await?;
check_scopes(&authed, || format!("runscript"))?;
if authed.is_operator {
return Err(error::Error::NotAuthorized(
"Operators cannot run preview jobs for security reasons".to_string(),
));
}
let mut job_id = None;
let mut tx = None;
let mut uploaded = false;
while let Some(field) = multipart.next_field().await.unwrap() {
let name = field.name().unwrap().to_string();
let data = field.bytes().await.unwrap();
if name == "preview" {
let preview: Preview = serde_json::from_slice(&data).map_err(to_anyhow)?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tag = run_query.tag.clone().or(preview.tag.clone());
check_tag_available_for_workspace(&w_id, &tag).await?;
let ltx =
PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq.clone());
let args = preview.args.unwrap_or_default();
// hmap.insert("")
let (uuid, ntx) = push(
&db,
ltx,
&w_id,
match preview.kind {
Some(PreviewKind::Identity) => JobPayload::Identity,
Some(PreviewKind::Noop) => JobPayload::Noop,
_ => JobPayload::Code(RawCode {
hash: Some(PREVIEW_IS_CODEBASE_HASH),
content: preview.content.unwrap_or_default(),
path: preview.path,
language: preview.language.unwrap_or(ScriptLang::Deno),
lock: preview.lock,
concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
concurrency_time_window_s: None, // TODO(gbouv): same as above
cache_ttl: None,
dedicated_worker: preview.dedicated_worker,
}),
},
args,
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
scheduled_for,
None,
None,
None,
run_query.job_id,
false,
false,
None,
true,
tag,
run_query.timeout,
None,
None,
)
.await?;
job_id = Some(uuid);
tx = Some(ntx);
}
if name == "file" {
let id = job_id
.as_ref()
.ok_or_else(|| {
Error::BadRequest(
"script need to be passed first in the multipart upload".to_string(),
)
})?
.to_string();
uploaded = true;
if let Some(os) = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS
.read()
.await
.clone()
{
let path = windmill_common::s3_helpers::bundle(&w_id, &id);
if let Err(e) = os
.put(&object_store::path::Path::from(path.clone()), data)
.await
{
tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e);
return Err(Error::ExecutionErr(format!("Failed to put {path} to s3")));
}
} else {
return Err(Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string()));
}
}
// println!("Length of `{}` is {} bytes", name, data.len());
}
if !uploaded {
return Err(Error::BadRequest("No file uploaded".to_string()));
}
if job_id.is_none() {
return Err(Error::BadRequest(
"No script found in the uploaded file".to_string(),
));
}
tx.unwrap().commit().await?;
Ok((StatusCode::CREATED, job_id.unwrap().to_string()))
}
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
async fn run_bundle_preview_script() -> error::Result<(StatusCode, String)> {
return Err(Error::BadRequest(
"bundle preview is an ee feature".to_string(),
));
}
#[derive(Deserialize)]
pub struct RunDependenciesRequest {
pub raw_scripts: Vec<RawScriptForDependencies>,
@@ -3649,84 +3484,6 @@ pub struct JobUpdate {
pub flow_status: Option<serde_json::Value>,
}
// #[cfg(all(feature = "enterprise", feature = "parquet"))]
// async fn get_logs_from_store(
// log_offset: i32,
// logs: &str,
// log_file_index: Option<Vec<String>>,
// ) -> Option<error::Result<Body>> {
// if log_offset > 0 {
// if let Some(file_index) = log_file_index {
// tracing::debug!("Getting logs from store: {file_index:?}");
// if let Some(os) = OBJECT_STORE_CACHE_SETTINGS.read().await.clone() {
// tracing::debug!("object store client present, streaming from there");
// let logs = logs.to_string();
// let stream = async_stream::stream! {
// for file_p in file_index {
// let file_p_2 = file_p.clone();
// let file = os.get(&object_store::path::Path::from(file_p)).await;
// if let Ok(file) = file {
// if let Ok(bytes) = file.bytes().await {
// yield Ok(bytes::Bytes::from(bytes)) as object_store::Result<bytes::Bytes>;
// }
// } else {
// tracing::debug!("error getting file from store: {file_p_2}: {}", file.err().unwrap());
// }
// }
// yield Ok(bytes::Bytes::from(logs))
// };
// return Some(Ok(Body::from_stream(stream)));
// } else {
// tracing::debug!("object store client not present, cannot stream logs from store");
// }
// }
// }
// return None;
// }
#[cfg(all(feature = "enterprise", feature = "parquet"))]
async fn get_log_file(Path((_w_id, file_p)): Path<(String, String)>) -> error::Result<Response> {
if let Some(os) = OBJECT_STORE_CACHE_SETTINGS.read().await.clone() {
let file = os
.get(&object_store::path::Path::from(format!("logs/{file_p}")))
.await;
if let Ok(file) = file {
if let Ok(bytes) = file.bytes().await {
use axum::http::header;
let res = Response::builder()
.header(header::CONTENT_TYPE, "text/plain")
.body(Body::from(bytes::Bytes::from(bytes)))
.unwrap();
return Ok(res);
} else {
return Err(error::Error::InternalErr(format!(
"Error getting bytes from file: {}",
file_p
)));
}
} else {
return Err(error::Error::NotFound(format!(
"File not found: {}",
file_p
)));
}
} else {
return Err(error::Error::InternalErr(
"Object store client not present, cannot stream logs from store".to_string(),
));
}
}
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
async fn get_log_file(Path((_w_id, file_p)): Path<(String, String)>) -> error::Result<Response> {
return Err(error::Error::NotFound(format!(
"Get log file is an EE feature: {}",
file_p
)));
}
async fn get_job_update(
Extension(db): Extension<DB>,
Path((w_id, job_id)): Path<(String, Uuid)>,
@@ -3883,17 +3640,6 @@ fn list_completed_jobs_query(
sqlb.and_where("result @> ?".bind(&result.replace("'", "''")));
}
if let Some(label) = &lq.label {
let mut wh = format!("result->'wm_labels' ? ");
wh.push_str(&format!("'{}'", &label.replace("'", "''")));
sqlb.and_where(&wh);
sqlb.and_where("result ? 'wm_labels'");
}
if lq.is_not_schedule.unwrap_or(false) {
sqlb.and_where("schedule_path IS null");
}
sqlb
}
#[derive(Deserialize, Clone)]
@@ -3925,8 +3671,6 @@ pub struct ListCompletedQuery {
pub scheduled_for_before_now: Option<bool>,
pub all_workspaces: Option<bool>,
pub has_null_parent: Option<bool>,
pub label: Option<String>,
pub is_not_schedule: Option<bool>,
}
async fn list_completed_jobs(
@@ -3974,7 +3718,6 @@ async fn list_completed_jobs(
"mem_peak",
"tag",
"priority",
"result->'wm_labels' as labels",
"'CompletedJob' as type",
],
)
@@ -3992,7 +3735,7 @@ async fn get_completed_job<'a>(
let job_o = sqlx::query("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path,
CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id,
schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped,
raw_lock, email, visible_to_owner, mem_peak, tag, priority, result->'wm_labels' as labels FROM completed_job WHERE id = $1 AND workspace_id = $2")
raw_lock, email, visible_to_owner, mem_peak, tag, priority FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(id)
.bind(w_id)
.fetch_optional(&db)

View File

@@ -25,7 +25,6 @@ use db::DB;
use git_version::git_version;
use reqwest::Client;
use std::collections::HashMap;
use std::time::Duration;
use std::{net::SocketAddr, sync::Arc};
use tokio::sync::RwLock;
use tower::ServiceBuilder;
@@ -105,8 +104,6 @@ lazy_static::lazy_static! {
pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new()
.user_agent("windmill/beta")
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(30))
.danger_accept_invalid_certs(std::env::var("ACCEPT_INVALID_CERTS").is_ok())
.build().unwrap();
@@ -144,12 +141,14 @@ pub async fn run_server(
));
let argon2 = Arc::new(Argon2::default());
let disable_response_logs = std::env::var("DISABLE_RESPONSE_LOGS")
.ok()
.map(|x| x == "true")
.unwrap_or(false);
let middleware_stack = ServiceBuilder::new()
.layer(
TraceLayer::new_for_http()
.on_response(MyOnResponse {})
.make_span_with(MyMakeSpan {})
.on_request(())
.on_failure(MyOnFailure {}),
)
.layer(Extension(db.clone()))
.layer(Extension(rsmq))
.layer(Extension(user_db))
@@ -286,18 +285,6 @@ pub async fn run_server(
.fallback(static_assets::static_handler)
.layer(middleware_stack);
let app = if disable_response_logs {
app
} else {
app.layer(
TraceLayer::new_for_http()
.on_response(MyOnResponse {})
.make_span_with(MyMakeSpan {})
.on_request(())
.on_failure(MyOnFailure {}),
)
};
let instance_name = rd_string(5);
let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
@@ -333,7 +320,6 @@ async fn is_up_to_date() -> Result<String, AppError> {
let error_reading_version = || anyhow::anyhow!("Error reading latest released version");
let version = HTTP_CLIENT
.get("https://api.github.com/repos/windmill-labs/windmill/releases/latest")
.timeout(Duration::from_secs(10))
.send()
.await
.context("Impossible to reach api.github")?

View File

@@ -221,7 +221,7 @@ async fn create_schedule(
.await?;
if ns.enabled.unwrap_or(true) {
tx = push_scheduled_job(&db, tx, &schedule).await?
tx = push_scheduled_job(&db, tx, schedule).await?
}
tx.commit().await?;
@@ -303,7 +303,7 @@ async fn edit_schedule(
.await?;
if schedule.enabled {
tx = push_scheduled_job(&db, tx, &schedule).await?;
tx = push_scheduled_job(&db, tx, schedule).await?;
}
tx.commit().await?;
@@ -512,7 +512,7 @@ pub async fn set_enabled(
.await?;
if payload.enabled {
tx = push_scheduled_job(&db, tx, &schedule).await?;
tx = push_scheduled_job(&db, tx, schedule).await?;
}
tx.commit().await?;
@@ -560,7 +560,7 @@ pub async fn set_enabled(
// .await?;
// if payload.enabled {
// tx = push_scheduled_job(&db, tx, &schedule).await?;
// tx = push_scheduled_job(&db, tx, schedule).await?;
// }
// tx.commit().await?;
@@ -581,37 +581,14 @@ async fn delete_schedule(
let path = path.to_path();
clear_schedule(&mut tx, path, &w_id).await?;
let exists = sqlx::query_scalar!(
"SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2",
sqlx::query!(
"DELETE FROM schedule WHERE path = $1 AND workspace_id = $2",
path,
w_id
)
.fetch_optional(&mut *tx)
.await?
.flatten();
if exists.is_none() {
return Err(windmill_common::error::Error::NotFound(format!(
"Schedule {} not found",
path
)));
}
let del = sqlx::query_scalar!(
"DELETE FROM schedule WHERE path = $1 AND workspace_id = $2 RETURNING 1",
path,
w_id
)
.fetch_optional(&mut *tx)
.await?
.flatten();
if del.is_none() {
return Err(windmill_common::error::Error::NotAuthorized(format!(
"Not authorized to delete schedule {}",
path
)));
}
.execute(&mut *tx)
.await?;
handle_deployment_metadata(
&authed.email,
@@ -800,7 +777,7 @@ pub async fn clear_schedule<'c>(
w_id: &str,
) -> Result<()> {
sqlx::query!(
"DELETE FROM queue WHERE schedule_path = $1 AND running = false AND workspace_id = $2 AND is_flow_step = false",
"DELETE FROM queue WHERE schedule_path = $1 AND running = false AND workspace_id = $2",
path,
w_id
)

View File

@@ -13,9 +13,6 @@ use crate::{
webhook_util::{WebhookMessage, WebhookShared},
HTTP_CLIENT,
};
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use axum::extract::Multipart;
use axum::{
extract::{Extension, Path, Query},
response::IntoResponse,
@@ -34,10 +31,6 @@ use std::{
};
use windmill_audit::audit_ee::audit_log;
use windmill_audit::ActionKind;
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use windmill_common::error::to_anyhow;
use windmill_common::{
db::UserDB,
error::{Error, JsonResult, Result},
@@ -98,8 +91,6 @@ pub struct ScriptWDraft {
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub no_main_func: Option<bool>,
}
pub fn global_service() -> Router {
@@ -123,7 +114,6 @@ pub fn workspaced_service() -> Router {
.route("/list", get(list_scripts))
.route("/list_search", get(list_search_scripts))
.route("/create", post(create_script))
.route("/create_snapshot", post(create_snapshot_script))
.route("/archive/p/*path", post(archive_script_by_path))
.route("/get/draft/*path", get(get_script_by_path_w_draft))
.route("/get/p/*path", get(get_script_by_path))
@@ -202,9 +192,7 @@ async fn list_scripts(
"tag",
"draft.path IS NOT NULL as has_draft",
"draft_only",
"ws_error_handler_muted",
"no_main_func",
"codebase IS NOT NULL as use_codebase"
"ws_error_handler_muted"
])
.left()
.join("favorite")
@@ -224,10 +212,6 @@ async fn list_scripts(
.limit(per_page)
.clone();
if authed.is_operator || lq.hide_without_main.unwrap_or(false) {
sqlb.and_where("o.no_main_func IS NOT TRUE");
}
if lq.show_archived.unwrap_or(false) {
sqlb.and_where_eq(
"o.created_at",
@@ -321,82 +305,6 @@ fn hash_script(ns: &NewScript) -> i64 {
dh.finish() as i64
}
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
async fn create_snapshot_script() -> Result<(StatusCode, String)> {
Err(Error::BadRequest("Upgrade to EE to use bundle".to_string()))
}
#[cfg(all(feature = "enterprise", feature = "parquet"))]
async fn create_snapshot_script(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
Extension(webhook): Extension<WebhookShared>,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
mut multipart: Multipart,
) -> Result<(StatusCode, String)> {
let mut script_hash = None;
let mut tx = None;
let mut uploaded = false;
while let Some(field) = multipart.next_field().await.unwrap() {
let name = field.name().unwrap().to_string();
let data = field.bytes().await.unwrap();
if name == "script" {
let ns = Some(serde_json::from_slice(&data).map_err(to_anyhow)?);
let (new_hash, ntx) = create_script_internal(
ns.unwrap(),
w_id.clone(),
authed.clone(),
db.clone(),
rsmq.clone(),
user_db.clone(),
webhook.clone(),
)
.await?;
script_hash = Some(new_hash.to_string());
tx = Some(ntx);
}
if name == "file" {
let hash = script_hash.as_ref().ok_or_else(|| {
Error::BadRequest(
"script need to be passed first in the multipart upload".to_string(),
)
})?;
uploaded = true;
if let Some(os) = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS
.read()
.await
.clone()
{
let path = windmill_common::s3_helpers::bundle(&w_id, &hash);
if let Err(e) = os
.put(&object_store::path::Path::from(path.clone()), data)
.await
{
tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e);
return Err(Error::ExecutionErr(format!("Failed to put {path} to s3")));
}
} else {
return Err(Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string()));
}
}
// println!("Length of `{}` is {} bytes", name, data.len());
}
if !uploaded {
return Err(Error::BadRequest("No file uploaded".to_string()));
}
if script_hash.is_none() {
return Err(Error::BadRequest(
"No script found in the uploaded file".to_string(),
));
}
tx.unwrap().commit().await?;
return Ok((StatusCode::CREATED, format!("{}", script_hash.unwrap())));
}
async fn create_script(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -406,24 +314,6 @@ async fn create_script(
Path(w_id): Path<String>,
Json(ns): Json<NewScript>,
) -> Result<(StatusCode, String)> {
let (hash, tx) = create_script_internal(ns, w_id, authed, db, rsmq, user_db, webhook).await?;
tx.commit().await?;
Ok((StatusCode::CREATED, format!("{}", hash)))
}
async fn create_script_internal<'c>(
ns: NewScript,
w_id: String,
authed: ApiAuthed,
db: sqlx::Pool<Postgres>,
rsmq: Option<rsmq_async::MultiplexedRsmq>,
user_db: UserDB,
webhook: WebhookShared,
) -> Result<(
ScriptHash,
QueueTransaction<'c, rsmq_async::MultiplexedRsmq>,
)> {
let codebase = ns.codebase.as_ref();
#[cfg(not(feature = "enterprise"))]
if ns.ws_error_handler_muted.is_some_and(|val| val) {
return Err(Error::BadRequest(
@@ -431,10 +321,12 @@ async fn create_script_internal<'c>(
.to_string(),
));
}
let script_path = ns.path.clone();
let hash = ScriptHash(hash_script(&ns));
let authed = maybe_refresh_folders(&ns.path, &w_id, authed, &db).await;
let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), user_db.begin(&authed).await?).into();
if sqlx::query_scalar!(
"SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2",
hash.0,
@@ -450,6 +342,7 @@ async fn create_script_internal<'c>(
.to_owned(),
));
};
let clashing_script = sqlx::query_as::<_, Script>(
"SELECT * FROM script WHERE path = $1 AND archived = false AND workspace_id = $2",
)
@@ -457,6 +350,7 @@ async fn create_script_internal<'c>(
.bind(&w_id)
.fetch_optional(&mut tx)
.await?;
struct ParentInfo {
p_hashes: Vec<i64>,
perms: serde_json::Value,
@@ -549,11 +443,13 @@ async fn create_script_internal<'c>(
r
}
}?;
let p_hashes = parent_hashes_and_perms.as_ref().map(|v| &v.p_hashes[..]);
let extra_perms = parent_hashes_and_perms
.as_ref()
.map(|v| v.perms.clone())
.unwrap_or(json!({}));
let lock = if !(ns.language == ScriptLang::Python3
|| ns.language == ScriptLang::Go
|| ns.language == ScriptLang::Bun
@@ -564,20 +460,23 @@ async fn create_script_internal<'c>(
ns.lock
.and_then(|e| if e.is_empty() { None } else { Some(e) })
};
let needs_lock_gen = lock.is_none();
let envs = ns.envs.as_ref().map(|x| x.as_slice());
let envs = if ns.envs.is_none() || ns.envs.as_ref().unwrap().is_empty() {
None
} else {
envs
};
//::text::json is to ensure we use serde_json with preserve order
sqlx::query!(
"INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \
content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30)",
delete_after_use, timeout, concurrency_key, visible_to_runner_only) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28)",
&w_id,
&hash.0,
ns.path,
@@ -606,11 +505,10 @@ async fn create_script_internal<'c>(
ns.timeout,
ns.concurrency_key,
ns.visible_to_runner_only,
ns.no_main_func,
codebase
)
.execute(&mut tx)
.await?;
let p_path_opt = parent_hashes_and_perms.as_ref().map(|x| x.p_path.clone());
if let Some(ref p_path) = p_path_opt {
sqlx::query!(
@@ -648,7 +546,7 @@ async fn create_script_internal<'c>(
clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?;
if schedule.enabled {
tx = push_scheduled_job(&db, tx, &schedule).await?;
tx = push_scheduled_job(&db, tx, schedule).await?;
}
}
} else {
@@ -660,6 +558,7 @@ async fn create_script_internal<'c>(
.execute(&mut tx)
.await?;
}
if p_hashes.is_some() && !p_hashes.unwrap().is_empty() {
audit_log(
&mut tx,
@@ -705,6 +604,7 @@ async fn create_script_internal<'c>(
},
);
}
let permissioned_as = username_to_permissioned_as(&authed.username);
if needs_lock_gen {
let tag = if ns.dedicated_worker.is_some_and(|x| x) {
@@ -753,7 +653,7 @@ async fn create_script_internal<'c>(
None,
)
.await?;
Ok((hash, new_tx))
new_tx.commit().await?;
} else {
handle_deployment_metadata(
&authed.email,
@@ -770,8 +670,10 @@ async fn create_script_internal<'c>(
false,
)
.await?;
Ok((hash, tx))
tx.commit().await?;
}
Ok((StatusCode::CREATED, format!("{}", hash)))
}
pub async fn get_hub_script_by_path(
@@ -822,7 +724,7 @@ async fn get_script_by_path_w_draft(
let mut tx = user_db.begin(&authed).await?;
let script_o = sqlx::query_as::<_, ScriptWDraft>(
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func FROM script LEFT JOIN draft ON
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only FROM script LEFT JOIN draft ON
script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script'
WHERE script.path = $1 AND script.workspace_id = $2 \
AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \

View File

@@ -21,17 +21,15 @@ use axum::{
Json, Router,
};
use mail_send::{mail_builder::MessageBuilder, SmtpClientBuilder};
use serde::Deserialize;
use tokio::time::timeout;
use windmill_common::{
error::{self, JsonResult, Result},
error::{self, to_anyhow, JsonResult, Result},
global_settings::{AUTOMATE_USERNAME_CREATION_SETTING, ENV_SETTINGS, HUB_BASE_URL_SETTING},
server::Smtp,
utils::send_email,
};
#[cfg(feature = "parquet")]
use windmill_common::error::to_anyhow;
pub fn global_service() -> Router {
#[warn(unused_mut)]
let r = Router::new()
@@ -69,17 +67,34 @@ pub async fn test_email(
require_super_admin(&db, &authed.email).await?;
let smtp = test_email.smtp;
let to = test_email.to;
let client_timeout = Duration::from_secs(3);
send_email(
"Test email from Windmill",
"Test email content",
vec![to],
smtp,
Some(client_timeout),
)
.await?;
let mut client = SmtpClientBuilder::new(smtp.host, smtp.port)
.implicit_tls(smtp.tls_implicit.unwrap_or(false));
if std::env::var("ACCEPT_INVALID_CERTS").is_ok() {
client = client.allow_invalid_certs();
}
let client = if let (Some(username), Some(password)) = (smtp.username, smtp.password) {
if !username.is_empty() {
client.credentials((username, password))
} else {
client
}
} else {
client
};
let message = MessageBuilder::new()
.from(("Windmill", smtp.from.as_str()))
.to(to.clone())
.subject("Test email from Windmill")
.text_body("Test email content");
let dur = Duration::from_secs(3);
timeout(dur, client.connect())
.await
.map_err(to_anyhow)?
.map_err(to_anyhow)?
.send(message)
.await
.map_err(to_anyhow)?;
tracing::info!("Sent test email to {to}");
Ok("Sent test email".to_string())
}

View File

@@ -32,6 +32,8 @@ use axum::{
};
use hyper::{header::LOCATION, StatusCode};
use lazy_static::lazy_static;
use mail_send::mail_builder::MessageBuilder;
use mail_send::SmtpClientBuilder;
use quick_cache::sync::Cache;
use rand::rngs::OsRng;
use regex::Regex;
@@ -44,11 +46,10 @@ use windmill_audit::audit_ee::audit_log;
use windmill_audit::ActionKind;
use windmill_common::global_settings::AUTOMATE_USERNAME_CREATION_SETTING;
use windmill_common::users::truncate_token;
use windmill_common::utils::send_email;
use windmill_common::worker::{CLOUD_HOSTED, SERVER_CONFIG};
use windmill_common::{
db::UserDB,
error::{self, Error, JsonResult, Result},
error::{self, to_anyhow, Error, JsonResult, Result},
users::SUPERADMIN_SECRET_EMAIL,
utils::{not_found_if_none, rd_string, require_admin, Pagination, StripPath},
};
@@ -1336,25 +1337,6 @@ async fn accept_invite(
.await?;
if let Some(r) = r {
let already_in_workspace = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM usr WHERE workspace_id = $1 AND email = $2)",
&nu.workspace_id,
&email,
)
.fetch_one(&mut *tx)
.await?
.unwrap_or(false);
if already_in_workspace {
tx.commit().await?;
return Ok((
StatusCode::CREATED,
format!(
"user {} accepted invite to workspace {}",
&email, nu.workspace_id
),
));
}
let username;
(tx, username) = add_user_to_workspace(
&nu.workspace_id,
@@ -1823,15 +1805,41 @@ pub fn send_email_if_possible(subject: &str, content: &str, to: &str) {
let content = content.to_string();
let to = to.to_string();
tokio::spawn(async move {
if let Err(e) = send_email_if_possible_intern(&subject, &content, to.clone()).await {
tracing::error!("Failed to send email to {}: {}", to, e);
if let Err(e) = send_email_if_possible_intern(&subject, &content, &to).await {
tracing::error!("Failed to send email to {}: {}", &to, e);
}
});
}
pub async fn send_email_if_possible_intern(subject: &str, content: &str, to: String) -> Result<()> {
pub async fn send_email_if_possible_intern(subject: &str, content: &str, to: &str) -> Result<()> {
if let Some(smtp) = SERVER_CONFIG.read().await.smtp.clone() {
send_email(subject, content, vec![to], smtp, None).await?;
let mut client = SmtpClientBuilder::new(smtp.host, smtp.port)
.implicit_tls(smtp.tls_implicit.unwrap_or(false));
if std::env::var("ACCEPT_INVALID_CERTS").is_ok() {
client = client.allow_invalid_certs();
}
let client = if let (Some(username), Some(password)) = (smtp.username, smtp.password) {
if !username.is_empty() {
client.credentials((username, password))
} else {
client
}
} else {
client
};
let message = MessageBuilder::new()
.from(("Windmill", smtp.from.as_str()))
.to(to)
.subject(subject)
.text_body(content);
client
.connect()
.await
.map_err(to_anyhow)?
.send(message)
.await
.map_err(to_anyhow)?;
tracing::info!("Sent email to {to}: {subject}");
}
return Ok(());
}

View File

@@ -71,7 +71,6 @@ impl WebhookShared {
let (tx, mut rx) = mpsc::unbounded_channel::<WebhookPayload>();
let _process = tokio::spawn(async move {
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(5))
// TODO: investigate pool timeouts and such if TCP load is high
.timeout(Duration::from_secs(5))
.build()

View File

@@ -14,16 +14,14 @@ use axum::{
use serde::{Deserialize, Serialize};
use sqlx::FromRow;
use uuid::Uuid;
use windmill_common::{
db::UserDB,
error::JsonResult,
utils::{paginate, Pagination},
worker::{ALL_TAGS, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE},
DB,
};
use crate::{db::ApiAuthed, utils::require_super_admin};
use crate::db::ApiAuthed;
pub fn global_service() -> Router {
Router::new()
@@ -35,7 +33,6 @@ pub fn global_service() -> Router {
get(get_default_tags_per_workspace),
)
.route("/get_default_tags", get(get_default_tags))
.route("/queue_metrics", get(get_queue_metrics))
}
#[derive(FromRow, Serialize, Deserialize)]
@@ -46,12 +43,9 @@ struct WorkerPing {
started_at: chrono::DateTime<chrono::Utc>,
ip: String,
jobs_executed: i32,
current_job_id: Option<Uuid>,
current_job_workspace_id: Option<String>,
custom_tags: Option<Vec<String>>,
worker_group: String,
wm_version: String,
occupancy_rate: Option<f32>,
}
#[derive(Serialize, Deserialize)]
@@ -68,25 +62,21 @@ pub struct ListWorkerQuery {
async fn list_worker_pings(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Query(query): Query<ListWorkerQuery>,
) -> JsonResult<Vec<WorkerPing>> {
let is_super_admin = require_super_admin(&db, &authed.email).await.is_ok();
let mut tx = user_db.begin(&authed).await?;
let (per_page, offset) = paginate(Pagination { page: query.page, per_page: query.per_page });
let rows = sqlx::query_as!(
WorkerPing,
"SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate
FROM worker_ping
WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)
ORDER BY ping_at desc LIMIT $2 OFFSET $3",
"SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags, worker_group, wm_version FROM worker_ping
WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)
ORDER BY ping_at desc LIMIT $2 OFFSET $3",
query.ping_since,
per_page as i64,
offset as i64,
is_super_admin
offset as i64
)
.fetch_all(&mut *tx)
.await?;
@@ -128,35 +118,3 @@ async fn get_default_tags_per_workspace() -> JsonResult<bool> {
async fn get_default_tags() -> JsonResult<Vec<String>> {
Ok(Json(DEFAULT_TAGS.clone()))
}
#[derive(Serialize)]
struct QueueMetric {
id: String,
values: Vec<serde_json::Value>,
}
async fn get_queue_metrics(
authed: ApiAuthed,
Extension(db): Extension<DB>,
) -> JsonResult<Vec<QueueMetric>> {
require_super_admin(&db, &authed.email).await?;
let queue_metrics = sqlx::query_as!(
QueueMetric,
"WITH queue_metrics as (
SELECT id, value, created_at
FROM metrics
WHERE id LIKE 'queue_%'
AND created_at > now() - interval '14 day'
ORDER BY created_at ASC
)
SELECT id, array_agg(json_build_object('value', value, 'created_at', created_at)) as \"values!\"
FROM queue_metrics
GROUP BY id
ORDER BY id ASC"
)
.fetch_all(&db)
.await?;
Ok(Json(queue_metrics))
}

View File

@@ -53,6 +53,7 @@ use windmill_common::{
variables::ExportableListableVariable,
};
use windmill_git_sync::handle_deployment_metadata;
use windmill_queue::QueueTransaction;
use crate::oauth2_ee::InstanceEvent;
use crate::variables::{decrypt, encrypt};
@@ -487,25 +488,25 @@ async fn run_slack_message_test_job(
json!(format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}")),
);
let uuid = windmill_queue::push_error_handler(
let tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into();
let (uuid, tx) = windmill_queue::handle_on_failure(
&db,
rsmq,
tx,
Uuid::parse_str("00000000-0000-0000-0000-000000000000")?,
None,
Some("slack_message_test".to_string()),
"slack_message_test",
"slack_message_test",
false,
w_id.as_str(),
&format!("script/{}", req.hub_script_path.as_str()),
sqlx::types::Json(&fake_result),
None,
Some(Utc::now()),
0,
Utc::now(),
Some(json!(extra_args)),
authed.email.as_str(),
false,
false,
None, // Note: we could mark it as high priority to return result quickly to the user
)
.await?;
tx.commit().await?;
Ok(Json(RunSlackMessageTestJobResponse {
job_uuid: uuid.to_string(),
@@ -721,16 +722,6 @@ async fn edit_auto_invite(
for user in users_to_auto_add.as_ref().unwrap() {
auto_add_user(&user.email, &w_id, &operator, &mut tx).await?;
send_email_if_possible(
&format!("Added to Windmill's workspace: {w_id}"),
&format!(
"You have been granted access to Windmill's workspace {w_id} by {email}.
Access the workspace at {}/?workspace={w_id}",
BASE_URL.read().await.clone()
),
&user.email,
);
}
} else {
sqlx::query!(
@@ -1945,22 +1936,6 @@ async fn invite_user(
let mut tx = db.begin().await?;
let already_in_workspace = sqlx::query_scalar!(
"SELECT EXISTS (SELECT 1 FROM usr WHERE workspace_id = $1 AND email = $2)",
&w_id,
nu.email
)
.fetch_one(&mut *tx)
.await?
.unwrap_or(false);
if already_in_workspace {
return Err(Error::BadRequest(format!(
"user with email {} already exists in workspace {}",
nu.email, w_id
)));
}
sqlx::query!(
"INSERT INTO workspace_invite
(workspace_id, email, is_admin, operator)
@@ -2071,14 +2046,6 @@ async fn add_user(
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM workspace_invite WHERE workspace_id = $1 AND email = $2",
&w_id,
nu.email
)
.execute(&mut *tx)
.await?;
sqlx::query_as!(
Group,
"INSERT INTO usr_to_group (workspace_id, usr, group_) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
@@ -2216,10 +2183,6 @@ struct ScriptMetadata {
pub restart_unless_cancelled: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub no_main_func: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub codebase: Option<String>,
}
pub fn is_none_or_false(val: &Option<bool>) -> bool {
@@ -2496,8 +2459,6 @@ async fn tarball_workspace(
delete_after_use: script.delete_after_use,
restart_unless_cancelled: script.restart_unless_cancelled,
visible_to_runner_only: script.visible_to_runner_only,
no_main_func: script.no_main_func,
codebase: script.codebase,
};
let metadata_str = serde_json::to_string_pretty(&metadata).unwrap();
archive

View File

@@ -10,7 +10,7 @@ enterprise = []
prometheus = ["dep:prometheus"]
flamegraph = ["dep:tracing-flame"]
loki = ["dep:tracing-loki"]
parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts", "dep:bytes"]
parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"]
[lib]
name = "windmill_common"
@@ -46,6 +46,4 @@ object_store = { workspace = true, optional = true }
prometheus = { workspace = true, optional = true }
aws-config = { workspace = true, optional = true }
aws-sdk-sts = { workspace = true, optional = true }
indexmap.workspace = true
bytes = { workspace = true, optional = true }
mail-send.workspace = true
indexmap.workspace = true

View File

@@ -1,5 +1,4 @@
use crate::ee::LicensePlan::Community;
use serde::Deserialize;
use std::sync::Arc;
use tokio::sync::RwLock;
@@ -19,9 +18,3 @@ pub async fn get_license_plan() -> LicensePlan {
// Implementation is not open source
return Community;
}
#[derive(Deserialize)]
#[serde(untagged)]
pub enum CriticalErrorChannel {}
pub async fn trigger_critical_error_channels(_error_message: String) {}

View File

@@ -8,7 +8,10 @@
use axum::body::Body;
use axum::response::Response;
use axum::{response::IntoResponse, response::Json};
use axum::{
response::IntoResponse,
response::Json,
};
use hyper::StatusCode;
use sqlx::migrate::MigrateError;
@@ -42,8 +45,6 @@ pub enum Error {
SqlErr(#[from] sqlx::Error),
#[error("Bad request: {0}")]
BadRequest(String),
#[error("Quota exceeded: {0}")]
QuotaExceeded(String),
#[error("Internal: {0}")]
InternalErr(String),
#[error("Hexadecimal decoding error: {0}")]
@@ -58,8 +59,6 @@ pub enum Error {
JsonErr(serde_json::Value),
#[error("{0}")]
OpenAIError(String),
#[error("{0}")]
AlreadyCompleted(String),
}
impl Error {
@@ -82,10 +81,9 @@ impl IntoResponse for Error {
Self::NotFound(_) => axum::http::StatusCode::NOT_FOUND,
Self::NotAuthorized(_) => axum::http::StatusCode::UNAUTHORIZED,
Self::RequireAdmin(_) => axum::http::StatusCode::FORBIDDEN,
Self::SqlErr(_)
| Self::BadRequest(_)
| Self::OpenAIError(_)
| Self::QuotaExceeded(_) => axum::http::StatusCode::BAD_REQUEST,
Self::SqlErr(_) | Self::BadRequest(_) | Self::OpenAIError(_) => {
axum::http::StatusCode::BAD_REQUEST
}
_ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
};

View File

@@ -13,19 +13,16 @@
use std::time::Duration;
pub async fn get_ip() -> anyhow::Result<String> {
tokio::select! {
biased;
_ = tokio::time::sleep(Duration::from_secs(10)) => {
return Err(anyhow::anyhow!("Expected to get ip under 10s"))
},
ip = reqwest::ClientBuilder::new()
.connect_timeout(Duration::from_secs(5))
.timeout(Duration::from_secs(5))
use reqwest::Result;
pub async fn get_ip() -> Result<String> {
reqwest::ClientBuilder::new()
.timeout(Duration::from_secs(3))
.build()?
.get("https://hub.windmill.dev/getip")
.send() => Ok(ip?
.error_for_status()?
.text().await?),
}
.send()
.await?
.error_for_status()?
.text()
.await
}

View File

@@ -29,7 +29,7 @@ pub fn is_retry_default(v: &RetryStatus) -> bool {
pub struct FlowStatus {
pub step: i32,
pub modules: Vec<FlowStatusModule>,
pub failure_module: Box<FlowStatusModuleWParent>,
pub failure_module: FlowStatusModuleWParent,
#[serde(skip_serializing_if = "HashMap::is_empty")]
#[serde(default)]
@@ -227,7 +227,7 @@ impl FlowStatus {
.iter()
.map(|m| FlowStatusModule::WaitingForPriorSteps { id: m.id.clone() })
.collect(),
failure_module: Box::new(FlowStatusModuleWParent {
failure_module: FlowStatusModuleWParent {
parent_module: None,
module_status: FlowStatusModule::WaitingForPriorSteps {
id: f
@@ -236,7 +236,7 @@ impl FlowStatus {
.map(|x| x.id.clone())
.unwrap_or_else(|| "failure".to_string()),
},
}),
},
cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
restarted_from: None,

View File

@@ -87,7 +87,7 @@ pub struct FlowValue {
pub modules: Vec<FlowModule>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
pub failure_module: Option<Box<FlowModule>>,
pub failure_module: Option<FlowModule>,
#[serde(default)]
#[serde(skip_serializing_if = "is_default")]
pub same_worker: bool,

View File

@@ -24,7 +24,6 @@ pub const OBJECT_STORE_CACHE_CONFIG_SETTING: &str = "object_store_cache_config";
pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation";
pub const HUB_BASE_URL_SETTING: &str = "hub_base_url";
pub const CRITICAL_ERROR_CHANNELS_SETTING: &str = "critical_error_channels";
pub const ENV_SETTINGS: [&str; 50] = [
"DISABLE_NSJAIL",

View File

@@ -243,8 +243,6 @@ pub struct CompletedJob {
pub tag: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
#[serde(skip_serializing_if = "Option::is_none")]
pub labels: Option<serde_json::Value>,
}
impl CompletedJob {
@@ -345,7 +343,6 @@ pub enum JobPayload {
pub struct RawCode {
pub content: String,
pub path: Option<String>,
pub hash: Option<i64>,
pub language: ScriptLang,
pub lock: Option<String>,
pub concurrent_limit: Option<i32>,

View File

@@ -11,7 +11,6 @@ use std::{
sync::{atomic::AtomicBool, Arc},
};
use ee::CriticalErrorChannel;
use error::Error;
use scripts::ScriptLang;
use sqlx::{Pool, Postgres};
@@ -45,7 +44,7 @@ pub mod workspaces;
pub mod tracing_init;
pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5;
pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 4;
pub const DEFAULT_HUB_BASE_URL: &str = "https://hub.windmill.dev";
@@ -74,9 +73,6 @@ lazy_static::lazy_static! {
pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
pub static ref HUB_BASE_URL: Arc<RwLock<String>> = Arc::new(RwLock::new(DEFAULT_HUB_BASE_URL.to_string()));
pub static ref CRITICAL_ERROR_CHANNELS: Arc<RwLock<Vec<CriticalErrorChannel>>> = Arc::new(RwLock::new(vec![]));
}
pub async fn shutdown_signal(

View File

@@ -177,48 +177,6 @@ pub fn build_object_store_client(
}
}
#[cfg(feature = "parquet")]
pub async fn attempt_fetch_bytes(
client: Arc<dyn ObjectStore>,
path: &str,
) -> error::Result<bytes::Bytes> {
use object_store::path::Path;
let object = client.get(&Path::from(path)).await;
if let Err(e) = object {
tracing::info!(
"Failed to pull bytes from object store at path {path}. Error: {:?}",
e
);
return Err(error::Error::ExecutionErr(format!(
"Failed to pull bytes from object store: {path}"
)));
}
let bytes = object.unwrap().bytes().await;
if bytes.is_err() {
tracing::info!(
"Failed to read bytes from object store: {path}. Error: {:?}",
bytes.err()
);
return Err(error::Error::ExecutionErr(format!(
"Failed to read bytes from object store: {path}"
)));
}
let bytes = bytes.unwrap();
tracing::info!("{path} len: {}", bytes.len());
if bytes.len() == 0 {
tracing::info!("object {path} not found in bucket, bytes empty",);
return Err(error::Error::ExecutionErr(format!(
"object {path} does not exist in bucket"
)));
}
return Ok(bytes);
}
#[cfg(feature = "parquet")]
use aws_config::{default_provider::credentials::DefaultCredentialsChain, Region};
#[cfg(feature = "parquet")]
@@ -450,7 +408,3 @@ impl CredentialProvider for AwsCredentialAdapter {
}))
}
}
pub fn bundle(w_id: &str, hash: &str) -> String {
format!("script_bundle/{}/{}", w_id, hash)
}

View File

@@ -128,8 +128,6 @@ impl Display for ScriptKind {
}
}
pub const PREVIEW_IS_CODEBASE_HASH: i64 = -42;
#[derive(Serialize, sqlx::FromRow)]
pub struct Script {
pub workspace_id: String,
@@ -177,10 +175,6 @@ pub struct Script {
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub no_main_func: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub codebase: Option<String>,
}
#[derive(Serialize, sqlx::FromRow)]
@@ -200,14 +194,6 @@ pub struct ListableScript {
pub draft_only: Option<bool>,
pub has_deploy_errors: bool,
pub ws_error_handler_muted: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub no_main_func: Option<bool>,
#[serde(skip_serializing_if = "is_false")]
pub use_codebase: bool,
}
fn is_false(x: &bool) -> bool {
return !x;
}
#[derive(Serialize)]
@@ -265,8 +251,6 @@ pub struct NewScript {
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
pub visible_to_runner_only: Option<bool>,
pub no_main_func: Option<bool>,
pub codebase: Option<String>,
}
fn lock_deserialize<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
@@ -336,7 +320,6 @@ pub struct ListScriptQuery {
pub is_template: Option<bool>,
pub kinds: Option<String>,
pub starred_only: Option<bool>,
pub hide_without_main: Option<bool>,
}
pub fn to_i64(s: &str) -> crate::error::Result<i64> {

View File

@@ -53,13 +53,7 @@ pub fn initialize_tracing() {
match json_fmt {
true => ts_base.with(json_layer().flatten_event(true)).init(),
false => ts_base
.with(
compact_layer()
.with_ansi(style.to_lowercase() != "never")
.with_file(true)
.with_line_number(true)
.with_target(false),
)
.with(compact_layer().with_ansi(style.to_lowercase() != "never"))
.init(),
}
}

View File

@@ -6,16 +6,11 @@
* LICENSE-AGPL for a copy of the license.
*/
#[cfg(feature = "enterprise")]
use crate::ee::trigger_critical_error_channels;
use crate::ee::LICENSE_KEY_ID;
use crate::error::{to_anyhow, Error, Result};
use crate::global_settings::UNIQUE_ID_SETTING;
use crate::server::Smtp;
use crate::DB;
use git_version::git_version;
use mail_send::mail_builder::MessageBuilder;
use mail_send::SmtpClientBuilder;
use rand::{distributions::Alphanumeric, thread_rng, Rng};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
@@ -90,20 +85,12 @@ pub async fn query_elems_from_hub(
url: &str,
query_params: Option<Vec<(&str, String)>>,
db: &DB,
) -> Result<(
reqwest::StatusCode,
reqwest::header::HeaderMap,
axum::body::Body,
)> {
) -> Result<(reqwest::StatusCode, reqwest::header::HeaderMap, axum::body::Body)> {
let response = http_get_from_hub(http_client, url, false, query_params, db).await?;
let status = response.status();
Ok((
status,
response.headers().clone(),
axum::body::Body::from_stream(response.bytes_stream()),
))
Ok((status, response.headers().clone(), axum::body::Body::from_stream(response.bytes_stream())))
}
pub async fn http_get_from_hub(
@@ -180,61 +167,3 @@ pub enum Mode {
Server,
Standalone,
}
pub async fn send_email(
subject: &str,
content: &str,
to: Vec<String>,
smtp: Smtp,
client_timeout: Option<tokio::time::Duration>,
) -> Result<()> {
let mut client = SmtpClientBuilder::new(smtp.host, smtp.port)
.implicit_tls(smtp.tls_implicit.unwrap_or(false));
if std::env::var("ACCEPT_INVALID_CERTS").is_ok() {
client = client.allow_invalid_certs();
}
let client = if let (Some(username), Some(password)) = (smtp.username, smtp.password) {
if !username.is_empty() {
client.credentials((username, password))
} else {
client
}
} else {
client
};
let message = MessageBuilder::new()
.from(("Windmill", smtp.from.as_str()))
.to(to.clone())
.subject(subject)
.text_body(content);
match client_timeout {
Some(timeout) => {
tokio::time::timeout(timeout, client.connect())
.await
.map_err(to_anyhow)?
.map_err(to_anyhow)?
.send(message)
.await
.map_err(to_anyhow)?;
}
None => {
client
.connect()
.await
.map_err(to_anyhow)?
.send(message)
.await
.map_err(to_anyhow)?;
}
}
tracing::info!("Sent email to {:#?}: {subject}", to);
return Ok(());
}
pub async fn report_critical_error(error_message: String) -> () {
tracing::error!("CRITICAL ERROR: {error_message}");
#[cfg(feature = "enterprise")]
trigger_critical_error_channels(error_message).await;
}

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