Compare commits

...

266 Commits

Author SHA1 Message Date
ben8t
0a121c6979 doc(task): improve TemplatedTask example 2025-12-19 14:06:42 +01:00
Aditya Kumar Puri
69a793b227 chore(executions): amend breadcrumb on the single execution page (#13544)
Closes https://github.com/kestra-io/kestra/issues/13394.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-12-11 11:02:52 +01:00
EM
35ccb3e39b refactor(core): remove usage of unnecessary i18n composable (#13580)
Closes https://github.com/kestra-io/kestra/issues/13351.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-11 10:51:53 +01:00
Piyush Bhaskar
3a7fcb2aa1 fix(core): changing Filters should reset to page 1 (#13596) 2025-12-11 14:39:11 +05:30
Miloš Paunović
103c5b92e9 chore(executions): show all available date options for the chart on the overview page (#13595)
Related to https://github.com/kestra-io/kestra/issues/13361.
2025-12-11 09:43:33 +01:00
Piyush Bhaskar
5253eeef95 fix(core): add seach where it was missed (#13594) 2025-12-11 14:08:14 +05:30
Miloš Paunović
848f835191 chore(core): properly load font weight 600 of public sans (#13593)
Closes https://github.com/kestra-io/kestra/issues/13592.
2025-12-11 09:33:23 +01:00
Malay Dewangan
3e55e67534 feat(plugin-metadata): add plugin metadata (#13539)
* docs(core-plugin-metadata): populate descriptions

---------

Co-authored-by: AJ Emerich <aj-emerich@proton.me>
2025-12-11 11:32:24 +05:30
Nicolas K.
7bca8b4924 fix(repositories): unwanted integer to string conversion (#13586)
* fix(repositories): unwanted integer to string conversion

* fix(repositories): clean code

---------

Co-authored-by: nKwiatkowski <nkwiatkowski@kestra.io>
2025-12-10 18:59:52 +01:00
Loïc Mathieu
56febfb415 fix(executions): don't remove worker task result for killed execution
As killing an executuion is asynchronous, it is inherently racy.
So when we kill an execution it move to the rerminal state which then will remove any worker task result to purge eagerly the queue table.

But if such worker task result arrives late and was not already processed by the executor, it will be purged before ever been able to be processed so the task would nevert be updated as KILLED.

Note: this may un-flaky some falky unit tests

Fixes https://github.com/kestra-io/kestra-ee/issues/6040
2025-12-10 17:27:11 +01:00
Nicolas K.
925b8c6954 fix(flows): deserialyze expression without cache (#13576)
* fix(flows): deserialyze expression without cache

* fix(flows): remove cache when deserialyzing expression

* fix(flows): remove cache when deserialyzing expression

---------

Co-authored-by: nKwiatkowski <nkwiatkowski@kestra.io>
2025-12-10 17:04:50 +01:00
mustafatarek
708816fe67 fix(core): return back to Collection check on yaml/json input values and add example of it in InputsTest 2025-12-10 16:31:14 +01:00
mustafatarek
5502473fa4 chore(core): change variable name to object 2025-12-10 16:31:14 +01:00
mustafatarek
c6cf0147a4 refactor(core): simplify changes by using variable 2025-12-10 16:31:14 +01:00
mustafatarek
2951f4b4bc feat(tests): add test coverage for json input type parsing as map 2025-12-10 16:31:14 +01:00
mustafatarek
4ea13e258b fix(core): fix parsing of json input type as for yaml 2025-12-10 16:31:14 +01:00
mustafatarek
3f8dcb47fd refactor(core): In case of yaml as map just return as it is, no need to serialize/deserialize again 2025-12-10 16:31:14 +01:00
mustafatarek
42dc3b930c fix(tests): pass previewInternalStorageFileFromExecution() test 2025-12-10 16:31:14 +01:00
mustafatarek
97a78abd28 refactor(core): update yaml test structure 2025-12-10 16:31:14 +01:00
mustafatarek
b3b2ef1b5a fix(tests): pass inputs() test 2025-12-10 16:31:14 +01:00
mustafatarek
596a26a137 fix(tests): pass all failed tests 2025-12-10 16:31:14 +01:00
mustafatarek
8a9a1df436 fix(tests): fix failed tests after adding another input 2025-12-10 16:31:14 +01:00
mustafatarek
55d0880ed3 refactor(tests): move test coverage to InputsTest instead of SubflowRunnerTest
- It is related to serializing inputs at resolving phase only
 - Added Inputs as Java Objects using yml should be serialized/deserialized properly to give the same structure at allValidInputs() test
2025-12-10 16:31:14 +01:00
mustafatarek
a74ebd5cd6 fix(tests): fix allValidTypedInputs() test by avoiding serialization of strings or other scalars, only serialize Maps and Collections 2025-12-10 16:31:14 +01:00
mustafatarek
f3aed38964 chore(tests): improve coding format at subflowInputTypeYmlSerialization() test 2025-12-10 16:31:14 +01:00
mustafatarek
2595e56199 feat(tests): add test coverage for subflow inputs serialized with type yaml 2025-12-10 16:31:14 +01:00
mustafatarek
e821bd7f65 refactor(core): use writeValueAsString() for all cases 2025-12-10 16:31:14 +01:00
mustafatarek
09762d2a8d fix(core): serialize subflow inputs with type yml properly 2025-12-10 16:31:14 +01:00
YannC
018c22918f Revert "feat(jdbc): method without auditlog registered for setting repository (#13543)" (#13581)
This reverts commit 3e9c8cf7da.
2025-12-10 16:00:55 +01:00
YannC
3e9c8cf7da feat(jdbc): method without auditlog registered for setting repository (#13543) 2025-12-10 15:52:42 +01:00
Nicolas K.
008404e442 feat(filters): add new operation to filters (#13424)
* feat(filters): add new operation to filters

* feat(filters): add flowId to flow search

* fix(test): fix unit test

---------

Co-authored-by: nKwiatkowski <nkwiatkowski@kestra.io>
2025-12-10 15:17:32 +01:00
Bhuvan C V
2b224bcde8 test(core): add regression test for illegal namespace updates
Adds a missing test case to FlowTest.java to verify that Flow.validateUpdate() correctly prevents namespace modification.
2025-12-10 14:36:00 +01:00
dependabot[bot]
1977b61693 build(deps): bump software.amazon.awssdk:bom from 2.40.0 to 2.40.5
Bumps software.amazon.awssdk:bom from 2.40.0 to 2.40.5.

---
updated-dependencies:
- dependency-name: software.amazon.awssdk:bom
  dependency-version: 2.40.5
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-10 14:31:32 +01:00
dependabot[bot]
8e2267f86c build(deps): bump com.fasterxml.uuid:java-uuid-generator
Bumps [com.fasterxml.uuid:java-uuid-generator](https://github.com/cowtowncoder/java-uuid-generator) from 5.1.1 to 5.2.0.
- [Commits](https://github.com/cowtowncoder/java-uuid-generator/compare/java-uuid-generator-5.1.1...java-uuid-generator-5.2.0)

---
updated-dependencies:
- dependency-name: com.fasterxml.uuid:java-uuid-generator
  dependency-version: 5.2.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-10 14:30:56 +01:00
Miloš Paunović
24355c2a88 refactor(executions): improve the trigger cascader on the overview page (#13524)
Closes https://github.com/kestra-io/kestra/issues/12942.
Closes https://github.com/kestra-io/kestra/issues/13283.
Closes https://github.com/kestra-io/kestra/issues/13290.
Closes https://github.com/kestra-io/kestra/issues/13294.
2025-12-10 13:08:16 +01:00
dependabot[bot]
51adcfa908 build(deps): bump flyingSaucerVersion from 10.0.5 to 10.0.6
Bumps `flyingSaucerVersion` from 10.0.5 to 10.0.6.

Updates `org.xhtmlrenderer:flying-saucer-core` from 10.0.5 to 10.0.6
- [Release notes](https://github.com/flyingsaucerproject/flyingsaucer/releases)
- [Changelog](https://github.com/flyingsaucerproject/flyingsaucer/blob/main/CHANGELOG.md)
- [Commits](https://github.com/flyingsaucerproject/flyingsaucer/compare/v10.0.5...v10.0.6)

Updates `org.xhtmlrenderer:flying-saucer-pdf` from 10.0.5 to 10.0.6
- [Release notes](https://github.com/flyingsaucerproject/flyingsaucer/releases)
- [Changelog](https://github.com/flyingsaucerproject/flyingsaucer/blob/main/CHANGELOG.md)
- [Commits](https://github.com/flyingsaucerproject/flyingsaucer/compare/v10.0.5...v10.0.6)

---
updated-dependencies:
- dependency-name: org.xhtmlrenderer:flying-saucer-core
  dependency-version: 10.0.6
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: org.xhtmlrenderer:flying-saucer-pdf
  dependency-version: 10.0.6
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-10 12:17:45 +01:00
dependabot[bot]
a55baa1f96 build(deps): bump software.amazon.awssdk.crt:aws-crt
Bumps [software.amazon.awssdk.crt:aws-crt](https://github.com/awslabs/aws-crt-java) from 0.40.1 to 0.40.3.
- [Release notes](https://github.com/awslabs/aws-crt-java/releases)
- [Commits](https://github.com/awslabs/aws-crt-java/compare/v0.40.1...v0.40.3)

---
updated-dependencies:
- dependency-name: software.amazon.awssdk.crt:aws-crt
  dependency-version: 0.40.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-10 12:17:21 +01:00
dependabot[bot]
32793fde18 build(deps): bump com.microsoft.playwright:playwright
Bumps [com.microsoft.playwright:playwright](https://github.com/microsoft/playwright-java) from 1.56.0 to 1.57.0.
- [Release notes](https://github.com/microsoft/playwright-java/releases)
- [Commits](https://github.com/microsoft/playwright-java/compare/v1.56.0...v1.57.0)

---
updated-dependencies:
- dependency-name: com.microsoft.playwright:playwright
  dependency-version: 1.57.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-10 12:15:37 +01:00
dependabot[bot]
4381d585ec build(deps): bump org.sonarqube from 7.1.0.6387 to 7.2.0.6526
Bumps org.sonarqube from 7.1.0.6387 to 7.2.0.6526.

---
updated-dependencies:
- dependency-name: org.sonarqube
  dependency-version: 7.2.0.6526
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-10 12:15:19 +01:00
dependabot[bot]
e595e26c45 build(deps): bump org.jooq:jooq from 3.20.9 to 3.20.10
Bumps org.jooq:jooq from 3.20.9 to 3.20.10.

---
updated-dependencies:
- dependency-name: org.jooq:jooq
  dependency-version: 3.20.10
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-10 12:15:00 +01:00
Miloš Paunović
b833cf28b5 chore(core): use the name property for tour flow element labels (#13569)
Closes https://github.com/kestra-io/kestra/issues/13546.
2025-12-10 10:46:36 +01:00
dependabot[bot]
ac11e9545c build(deps-dev): bump @types/node in /ui in the types group (#13564)
Bumps the types group in /ui with 1 update: [@types/node](https://github.com/DefinitelyTyped/DefinitelyTyped/tree/HEAD/types/node).


Updates `@types/node` from 24.10.1 to 24.10.2
- [Release notes](https://github.com/DefinitelyTyped/DefinitelyTyped/releases)
- [Commits](https://github.com/DefinitelyTyped/DefinitelyTyped/commits/HEAD/types/node)

---
updated-dependencies:
- dependency-name: "@types/node"
  dependency-version: 24.10.2
  dependency-type: direct:development
  update-type: version-update:semver-patch
  dependency-group: types
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-12-10 08:54:02 +01:00
dependabot[bot]
a07df5f6cd build(deps): bump the build group in /ui with 3 updates (#13563)
Bumps the build group in /ui with 3 updates: [@esbuild/darwin-arm64](https://github.com/evanw/esbuild), [@esbuild/darwin-x64](https://github.com/evanw/esbuild) and [@esbuild/linux-x64](https://github.com/evanw/esbuild).


Updates `@esbuild/darwin-arm64` from 0.27.0 to 0.27.1
- [Release notes](https://github.com/evanw/esbuild/releases)
- [Changelog](https://github.com/evanw/esbuild/blob/main/CHANGELOG.md)
- [Commits](https://github.com/evanw/esbuild/compare/v0.27.0...v0.27.1)

Updates `@esbuild/darwin-x64` from 0.27.0 to 0.27.1
- [Release notes](https://github.com/evanw/esbuild/releases)
- [Changelog](https://github.com/evanw/esbuild/blob/main/CHANGELOG.md)
- [Commits](https://github.com/evanw/esbuild/compare/v0.27.0...v0.27.1)

Updates `@esbuild/linux-x64` from 0.27.0 to 0.27.1
- [Release notes](https://github.com/evanw/esbuild/releases)
- [Changelog](https://github.com/evanw/esbuild/blob/main/CHANGELOG.md)
- [Commits](https://github.com/evanw/esbuild/compare/v0.27.0...v0.27.1)

---
updated-dependencies:
- dependency-name: "@esbuild/darwin-arm64"
  dependency-version: 0.27.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@esbuild/darwin-x64"
  dependency-version: 0.27.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@esbuild/linux-x64"
  dependency-version: 0.27.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-12-10 08:53:46 +01:00
Bikash Agarwala
f626c85346 fix: allow zero-byte file uploads in execution inputs (fixes #8218) 2025-12-09 12:52:25 +01:00
Roman Acevedo
e15b53ebb5 ci: add missing GH_PERSONAL_TOKEN in main-build.yml 2025-12-09 10:55:43 +01:00
Roman Acevedo
7edb6bc379 fix(blueprints): bring back up edit button
- fixes https://github.com/kestra-io/kestra-ee/issues/6070
2025-12-09 10:45:27 +01:00
Roman Acevedo
78c81f932b ci: add GH_PERSONAL_TOKEN in release-docker.yml CI for helm chart 2025-12-09 10:14:33 +01:00
Shankar
56bb3ca29c Fix week format in filter 2025-12-09 10:00:37 +01:00
Loïc Mathieu
14029e8c14 chore(tests): isolate concurrency related tests in their own class 2025-12-09 09:57:49 +01:00
char zheng
bea3d63d89 fix(executions): concurrency limit exceeded for KILLED execution
Fixes #13211
2025-12-09 09:57:49 +01:00
Shivansh Sharma
24a3bbd303 11229 : Ensure that a label key cannot contain spaces, special characters or encoded values (e.g. emojis) 2025-12-09 09:32:22 +01:00
Nirnay
f9932af2e8 fix(ui): Updated the trigger button (#13521) 2025-12-09 13:56:24 +05:30
Avirup Banik
e0410c8f24 made system overview page responsive (#13527) 2025-12-09 12:10:25 +05:30
Loïc Mathieu
424a6cb41a fix(execution): skip the render cache in flowable for properties used to compute next tasks
As when the flowable is itself in a flowable that process tasks concurrently like the ForEach when using a concurrency limit, it can be done multiple time with different values.

This can only occurs if the expression is using `taskRun.value`.

Fixes https://github.com/kestra-io/kestra-ee/issues/6055
2025-12-08 15:03:06 +01:00
Debjyoti Shit
afde71e913 fix(core): skip login screen after initial setup and send to welcome (#13489)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
Co-authored-by: Piyush Bhaskar <impiyush0012@gmail.com>
2025-12-08 18:09:50 +05:30
Miloš Paunović
086c32e711 chore(flows): redirect with applied filters from the overview page (#13522)
Closes https://github.com/kestra-io/kestra/issues/13392.
2025-12-08 10:52:46 +01:00
github-actions[bot]
710abcfaac chore(core): localize to languages other than english (#13520)
Co-authored-by: GitHub Action <actions@github.com>
2025-12-08 14:26:03 +05:30
Kavyakapoor
be951d015c fix(core): make password requirement descriptive. (#13483) 2025-12-08 14:15:11 +05:30
Piyush Bhaskar
a07260bef4 fix(core): refine navigation for authentication and setup routes (#13517) 2025-12-08 14:13:28 +05:30
Piyush Bhaskar
dd19f8391d chore(version): bump ui-libs (#13518) 2025-12-08 14:11:59 +05:30
mustafatarek
354873e220 chore(core): remove unnecessary attempt list copying 2025-12-08 09:41:10 +01:00
luoxin
386d4a15f0 fix(system): enable parallel loading for namespace files. (#13375)
* perf(core): enable parallel loading for namespace files.

* refactor(core): extract thread count calculation to avoid duplication.

* resolve namespaceFilesWithNamespaces test error.

---------

Co-authored-by: luoxin5 <luoxin5@xiaomi.com>
2025-12-08 09:23:35 +01:00
Yaswanth B
1b75f15680 refactor(core): remove usage of unnecessary i18n composable (#13492)
Closes https://github.com/kestra-io/kestra/issues/13352.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-08 08:42:56 +01:00
Richard-Mackey
957bf74d97 fix(core): make menuCollapsed = true on small screen (#13238)
Co-authored-by: Piyush Bhaskar <impiyush0012@gmail.com>
2025-12-06 02:58:37 +05:30
Florian Hussonnois
3cbad1ce0d fix(tests): fix StatefulTriggerInterfaceTest 2025-12-05 17:56:32 +01:00
Debjyoti Shit
760050e9fc fix(ui): improve responsive layout and styling of flow editor and sidebar (#13371)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
Co-authored-by: Bart Ledoux <bledoux@kestra.io>
2025-12-05 17:35:25 +01:00
Nancy Sangani
43f47ec337 fix: error dialogue not appearing after first use of checks #13357 (#13364)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-12-05 16:24:53 +01:00
Avirup Banik
ad5521199a Fix(ui): button glow timing and banner placement (#13417) 2025-12-05 18:44:22 +05:30
brian-mulier-p
fe7849d7fe fix(metadata): add system namespace to migrated namespaces (#13419)
closes https://github.com/kestra-io/kestra-ee/issues/6019
2025-12-05 14:01:02 +01:00
Loïc Mathieu
feeaeff0b2 chore(tests): remove running tests in parallel 2025-12-05 13:23:29 +01:00
Nirnay
ed6bc50163 fix(ui): Overflow of Flow labels in flow editor page (#13374) 2025-12-05 12:30:53 +01:00
Pratik Dey
069845f579 chore(core): properly handle conditional visibility of elements on dependency view (#13387)
Closes https://github.com/kestra-io/kestra/issues/13291.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-12-05 11:33:48 +01:00
Aditya Kumar Puri
f613eb0433 refactor(core): remove usage of unnecessary i18n composable (#13389)
Closes https://github.com/kestra-io/kestra/issues/13201.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-05 09:55:18 +01:00
Florian Hussonnois
e440c402b4 fix(flows): add flow variables to runContext for checks
Related-to: kestra-io/kestra-ee#5759
2025-12-04 16:13:32 +01:00
Miloš Paunović
700527b5dc chore(executions): properly cast values for the disabled property to boolean (#13384) 2025-12-04 15:50:32 +01:00
Miloš Paunović
5245014a32 chore(executions): avoid uppercase letters and monospace fonts for property labels (#13383)
Closes https://github.com/kestra-io/kestra/issues/13363.
2025-12-04 13:56:50 +01:00
Kollakota Siva Sai
5db0f44fb6 refactor(core): remove usage of unnecessary i18n composable (#13382)
Closes https://github.com/kestra-io/kestra/issues/13353.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-04 13:37:05 +01:00
brian-mulier-p
a8635108b7 fix(core): safeguard for null flow when trying to reset trigger in JdbcExecutor (#13381) 2025-12-04 12:46:53 +01:00
Miloš Paunović
cd4470044e feat(executions): make prev/next buttons loop through the executions of that flow (#13296)
Closes https://github.com/kestra-io/kestra/issues/9873.
2025-12-04 12:28:24 +01:00
yuri1969
4ec7f23a7b Amend inconsistency 2025-12-04 12:20:16 +01:00
yuri1969
107ba16ce3 feat(core): modernize Pebble cache
* Modernized the LRU cache from Guava to Caffeine.
* Registered metrics.
2025-12-04 12:20:16 +01:00
Miloš Paunović
042d548598 refactor(core): remove all traces of the old e2e setup (#13356) 2025-12-04 12:12:59 +01:00
Kunal
94bd6f0a1e chore(executions): amend alignment of the timeline element on the overview page (#13370)
Closes https://github.com/kestra-io/kestra/issues/13282.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-12-04 11:57:56 +01:00
Avirup Banik
f43f11e125 refactor(core): remove usage of unnecessary i18n composable (#13379)
Closes https://github.com/kestra-io/kestra/issues/13354.

Signed-off-by: Avirup Banik <avirup.banik2017@gmail.com>
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-12-04 11:23:59 +01:00
brian-mulier-p
3dfa5f97c4 fix(core): deprecate Await util (#13369)
This reverts commit 9fa94deba9.
2025-12-04 09:57:12 +01:00
Roman Acevedo
8898ba736b test(executions): fix ES duration tests 2025-12-03 18:37:49 +01:00
Loïc Mathieu
f5665bf719 Update core/src/test/java/io/kestra/core/serializers/YamlParserTest.java 2025-12-03 17:34:52 +01:00
Loïc Mathieu
b6db688003 Update core/src/test/resources/flows/valids/labels-deserialization.yaml 2025-12-03 17:34:52 +01:00
lizi3
93f5e366ed fix: add unit test 2025-12-03 17:34:52 +01:00
lizi3
0465ffa5df fix: unify label deserialization for numeric values in array form
The START_ARRAY branch did not apply type conversion for label values,
causing Integer/Boolean values to fail during execution state update.
Applying the same allowed-type conversion logic as the START_OBJECT branch
fixes the inconsistency and prevents cast errors.
2025-12-03 17:34:52 +01:00
brian.mulier
e869c54883 fix(ns-files): prevent ns files revision history failure & working restore
closes https://github.com/kestra-io/kestra-ee/issues/6022
2025-12-03 16:04:39 +01:00
YannC
32da15b2ea fix: missing tenant id for flow creation in AbstractSchedulerTest (#13311) 2025-12-03 15:58:52 +01:00
François Delbrayelle
a72ecfc2eb chore(icons): remove white backgrounds (#13362) 2025-12-03 15:27:24 +01:00
sarika
7cb494b244 refactor(core): remove usage of unnecessary i18n composable (#13366)
Closes https://github.com/kestra-io/kestra/issues/13224.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-03 15:19:10 +01:00
brian.mulier
9d21ab4b26 fix(cli): don't throw on ns files migration for ns without files
closes https://github.com/kestra-io/kestra-ee/issues/6019#event-21325612781
2025-12-03 14:32:30 +01:00
brian.mulier
8f3a5058b1 fix(cli): also fetch parent namespaces resources for metadata migration
part of https://github.com/kestra-io/kestra-ee/issues/6019#event-21325612781
2025-12-03 14:32:30 +01:00
Loïc Mathieu
56fb304ff6 fix(execution): NORMAL kind should also be retrieved
Fixes #13262
2025-12-03 11:43:50 +01:00
dependabot[bot]
28370d80df build(deps): bump dev.langchain4j:langchain4j-community-bom
Bumps [dev.langchain4j:langchain4j-community-bom](https://github.com/langchain4j/langchain4j-community) from 1.8.0-beta15 to 1.9.1-beta17.
- [Release notes](https://github.com/langchain4j/langchain4j-community/releases)
- [Commits](https://github.com/langchain4j/langchain4j-community/compare/1.8.0-beta15...1.9.1-beta17)

---
updated-dependencies:
- dependency-name: dev.langchain4j:langchain4j-community-bom
  dependency-version: 1.9.1-beta17
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-03 11:13:13 +01:00
dependabot[bot]
cc5fd30b2c build(deps): bump software.amazon.awssdk:bom from 2.39.4 to 2.40.0
Bumps software.amazon.awssdk:bom from 2.39.4 to 2.40.0.

---
updated-dependencies:
- dependency-name: software.amazon.awssdk:bom
  dependency-version: 2.40.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-03 10:45:32 +01:00
dependabot[bot]
61c39d23c5 build(deps): bump dev.langchain4j:langchain4j-bom from 1.8.0 to 1.9.1
Bumps [dev.langchain4j:langchain4j-bom](https://github.com/langchain4j/langchain4j) from 1.8.0 to 1.9.1.
- [Release notes](https://github.com/langchain4j/langchain4j/releases)
- [Commits](https://github.com/langchain4j/langchain4j/compare/1.8.0...1.9.1)

---
updated-dependencies:
- dependency-name: dev.langchain4j:langchain4j-bom
  dependency-version: 1.9.1
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-03 10:45:21 +01:00
dependabot[bot]
b5a40b2fcc build(deps): bump com.azure:azure-sdk-bom from 1.3.2 to 1.3.3
Bumps [com.azure:azure-sdk-bom](https://github.com/azure/azure-sdk-for-java) from 1.3.2 to 1.3.3.
- [Release notes](https://github.com/azure/azure-sdk-for-java/releases)
- [Commits](https://github.com/azure/azure-sdk-for-java/compare/azure-sdk-bom_1.3.2...azure-sdk-bom_1.3.3)

---
updated-dependencies:
- dependency-name: com.azure:azure-sdk-bom
  dependency-version: 1.3.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-12-03 10:43:57 +01:00
Miloš Paunović
825b9dbcdb build(deps): improve storybook related grouping of dependabot pull requests (#13344) 2025-12-03 09:21:36 +01:00
YannC
393b132444 fix: correct regex when importing flow (#13320) 2025-12-03 09:06:18 +01:00
Miloš Paunović
b0ce760e50 build(deps): improve grouping of dependabot pull requests for npm ecosystem (#13337) 2025-12-03 08:53:52 +01:00
dependabot[bot]
2d68dad70c build(deps): bump the build group in /ui with 3 updates (#13321)
Bumps the build group in /ui with 3 updates: [@swc/core-darwin-arm64](https://github.com/swc-project/swc), [@swc/core-darwin-x64](https://github.com/swc-project/swc) and [@swc/core-linux-x64-gnu](https://github.com/swc-project/swc).


Updates `@swc/core-darwin-arm64` from 1.15.2 to 1.15.3
- [Release notes](https://github.com/swc-project/swc/releases)
- [Changelog](https://github.com/swc-project/swc/blob/main/CHANGELOG.md)
- [Commits](https://github.com/swc-project/swc/compare/v1.15.2...v1.15.3)

Updates `@swc/core-darwin-x64` from 1.15.2 to 1.15.3
- [Release notes](https://github.com/swc-project/swc/releases)
- [Changelog](https://github.com/swc-project/swc/blob/main/CHANGELOG.md)
- [Commits](https://github.com/swc-project/swc/compare/v1.15.2...v1.15.3)

Updates `@swc/core-linux-x64-gnu` from 1.15.2 to 1.15.3
- [Release notes](https://github.com/swc-project/swc/releases)
- [Changelog](https://github.com/swc-project/swc/blob/main/CHANGELOG.md)
- [Commits](https://github.com/swc-project/swc/compare/v1.15.2...v1.15.3)

---
updated-dependencies:
- dependency-name: "@swc/core-darwin-arm64"
  dependency-version: 1.15.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@swc/core-darwin-x64"
  dependency-version: 1.15.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@swc/core-linux-x64-gnu"
  dependency-version: 1.15.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-12-03 08:25:18 +01:00
Loïc Mathieu
d08a2d8930 fix(executions): support Download content dispositions with brackets
By escaping them with %5B and %5D.

Fixes #13299
2025-12-02 16:02:02 +01:00
Vinicius Wrubleski
2722735d2d refactor(core): remove usage of unnecessary i18n composable (#13305)
Closes https://github.com/kestra-io/kestra/issues/13259.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-02 15:01:37 +01:00
Debjyoti Shit
e9b7d190d4 refactor(core): remove usage of unnecessary i18n composable (#13298)
Closes https://github.com/kestra-io/kestra/issues/13260.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-02 14:27:38 +01:00
Suraj Bhandarkar S
d6cfa01fd5 refactor(core): remove usage of unnecessary i18n composable (#13302)
Closes https://github.com/kestra-io/kestra/issues/13227.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-02 14:24:49 +01:00
brian-mulier-p
d8e3a9bd44 fix(executions): avoid infinite loop in some cases of execution failure (#13293) 2025-12-02 13:09:34 +01:00
brian-mulier-p
b18e3b76ef refacto(core): expose TaskLogLineMatcher (#13300) 2025-12-02 12:46:47 +01:00
Miloš Paunović
4660091dc9 chore(executions): amend label color on overview page for both themes (#13295)
Closes https://github.com/kestra-io/kestra/issues/13289.
2025-12-02 12:10:37 +01:00
Miloš Paunović
067414ffbe chore(core): enhance github issue templates by adding type (#13292) 2025-12-02 12:09:48 +01:00
brian.mulier
26f6154eed fix(tests): NOT EQUALS on NamespaceFileMetadataRepositoryTest query was too flaky 2025-12-02 11:46:52 +01:00
Loïc Mathieu
ea44128d2b chore(system): refactor RunnerUtils to be a static utils
Part-of: https://github.com/kestra-io/kestra-ee/issues/4228
2025-12-02 10:37:46 +01:00
Loïc Mathieu
4602546045 chore(system): refactor the TaskLogLineMatcher as a standard class
https://github.com/kestra-io/kestra-ee/issues/4228
2025-12-02 10:37:16 +01:00
Deepika Vaddevalli
aecd050314 refactor(core): remove usage of unnecessary i18n composable (#13286)
Closes https://github.com/kestra-io/kestra/issues/13258.

Signed-off-by: deepika1214 <deepikav201818@gmail.com>
Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-02 10:34:18 +01:00
Pa1
d6c290cb91 fix: documentation pannel overflow (#13230)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
Co-authored-by: Bart Ledoux <bledoux@kestra.io>
2025-12-02 10:22:12 +01:00
Barthélémy Ledoux
56216ef0b4 fix: loding of icons should be done with resources not in layout (#13288) 2025-12-02 10:21:20 +01:00
Loïc Mathieu
e1f983cc2d chore(system): refacto TreadMainFactoryBuilder to be a static utility
Part-of: https://github.com/kestra-io/kestra-ee/issues/4228
2025-12-02 10:05:20 +01:00
Loïc Mathieu
68f92e1159 chore(executions): replace usage of the DefaultRunContext by supported API
Part-of: https://github.com/kestra-io/kestra-ee/issues/4228
2025-12-02 10:04:24 +01:00
kkash08
5b597b9520 Fix ZIP download so that file extension remains .yaml 2025-12-02 09:24:55 +01:00
github-actions[bot]
b0606a4380 chore(core): localize to languages other than english (#13280)
Extended localization support by adding translations for multiple languages using English as the base. This enhances accessibility and usability for non-English-speaking users while keeping English as the source reference.

Co-authored-by: GitHub Action <actions@github.com>
2025-12-02 08:47:30 +01:00
Miloš Paunović
06450bfd65 feat(executions): redesign the overview page (#13093)
Closes https://github.com/kestra-io/kestra/issues/10780.
Closes https://github.com/kestra-io/kestra/issues/12749.
2025-12-02 08:43:21 +01:00
Piyush Bhaskar
ce12e19f99 fix(filters): bring back the removed visibledefault on reset (#13273) 2025-12-02 12:53:42 +05:30
Nirnay
e20d67f4f2 feat(ui): Search bar in Plugins breadcrumb (#13250) 2025-12-02 12:13:09 +05:30
brian.mulier
4d353937c3 feat(ns-files): introduce PurgeFiles task
part of #5617
2025-12-01 18:13:49 +01:00
brian.mulier
8edad60695 feat(ns-files): introduce revision history
closes https://github.com/kestra-io/kestra/issues/5617
2025-12-01 18:13:49 +01:00
brian.mulier
c0ecc2cb20 refacto(revisions): introduce a generic Revisions component 2025-12-01 18:13:49 +01:00
brian.mulier
682d258e7b feat(ns-files): add a metadata layer on top for better performance & versioned ns files
part of https://github.com/kestra-io/kestra/issues/5617
2025-12-01 18:13:49 +01:00
brian.mulier
d20f7039c7 fix(tests): avoid flakiness with random worker group in JdbcServiceLivenessCoordinatorTest 2025-12-01 16:56:01 +01:00
brian.mulier
4e1b53fadf fix(tests): add tenant to ConcurrencyLimitServiceTest 2025-12-01 16:56:01 +01:00
brian.mulier
2191331750 fix(tests): JdbcRunnerRetryTest.retryFailedFlowDuration flaky 2025-12-01 16:56:01 +01:00
brian.mulier
90c3281eae fix(tests): WorkingDirectoryTest.outputFiles flaky 2025-12-01 16:56:01 +01:00
brian.mulier
9fa94deba9 refacto(core): rename Await.until(sleep) and (timeout) to avoid confusions 2025-12-01 16:56:01 +01:00
brian.mulier
9d73d72ab0 fix(tests): ensure Executor is running before proceeding 2025-12-01 16:56:01 +01:00
YannC
4799ee320f fix: Remove per namespace dashboard on default flow dashboard (#13218)
close #12841

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-12-01 16:07:37 +01:00
Roman Acevedo
40880bf7d8 feat(system): force nullsFirst for ASC and opposite globally in JDBC 2025-12-01 16:06:35 +01:00
Roman Acevedo
7aca309be5 feat(dashboard): calculate running execution duration on the fly 2025-12-01 16:06:35 +01:00
Roman Acevedo
64899f3103 fix(dashboard): avoid undefined Date being mapped to current date 2025-12-01 16:06:35 +01:00
Roman Acevedo
fbe6df34ca feat(executions): set duration to null for non-terminated execs and fix frontend duration 2025-12-01 16:06:35 +01:00
Nirnay
df21ef4064 fix(ui): scrollbar flickering (#13263)
Co-authored-by: Piyush Bhaskar <impiyush0012@gmail.com>
2025-12-01 17:51:29 +05:30
Loïc Mathieu
64a2c3b746 chore(system): add @CheckReturnValue to AclCheck methods to be sure check() is called 2025-12-01 12:45:05 +01:00
brian-mulier-p
f06b1c5347 fix(core): concurrency limit on JDBC was decrementing when using FAIL or CANCEL behavior (#13220)
closes https://github.com/kestra-io/kestra/issues/13141
2025-12-01 12:44:30 +01:00
brian-mulier-p
ef154bb029 fix(core): acl.check was not called (#13265) 2025-12-01 12:09:34 +01:00
Loïc Mathieu
496e01eb3e feat(executions): add support to ACL check inside the run context
Part-of: https://github.com/kestra-io/kestra-ee/issues/4228
2025-12-01 10:29:38 +01:00
Piyush Bhaskar
f2c15185fb feat(filters): add default visible filters for state and level (#13255) 2025-12-01 14:23:30 +05:30
Loïc Mathieu
20c5328199 feat(system): deprecate RunContext method that must not be used anymore
- `isInitialized()` is not used by plugin so should not be part of the interface
- `getApplicationContext` should not be used anymore, all tasks that uses it would be refactored to avoid its usage

Part-of: https://github.com/kestra-io/kestra-ee/issues/4228
2025-12-01 09:51:12 +01:00
Loïc Mathieu
91330496f2 fix(execution): failed flowable should also have a failed attempt
When a flowable fail, it should also switch its attempts to FAILED.

Fixes #12614
2025-12-01 09:50:38 +01:00
Georg Traar
101700ac53 chore(dashboards): display in progress/pending executions as absolute values (#13237)
Running (technically "In Progress") and pending executions were previously displayed as ratios of total executions,
which seemed to convey little value. This update switches
these KPIs to absolute values to provide a clearer, more intuitive view of current system load and renames "Running" to "In Progress" in line with other charts on the dashboards.

Co-authored-by: Georg Traar <georg@crate.io>
2025-12-01 09:35:40 +01:00
Ronin@73
36389d7d79 chore(executions): prevent tooltip from being stuck on gantt view (#13209)
Closes https://github.com/kestra-io/kestra/issues/12734.

Co-authored-by: AtulRaghuvanshi73 <atul.raghuvanshi73@gmail.com>
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-12-01 09:11:57 +01:00
Eshaan Gupta
f29dbe53a8 refactor(core): remove usage of unnecessary i18n composable (#13246)
Closes https://github.com/kestra-io/kestra/issues/13228.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-01 08:58:31 +01:00
Pratik Dey
a6d34151bf chore(core): remove unnecessary button in backfill dialog (#13249)
Closes https://github.com/kestra-io/kestra/issues/13184.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-01 08:52:54 +01:00
Nancy Sangani
4e54fac980 refactor(core): remove usage of unnecessary i18n composable (#13252)
Closes https://github.com/kestra-io/kestra/issues/13226.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-12-01 08:48:14 +01:00
github-actions[bot]
6e50654544 chore(core): localize to languages other than english (#13254)
Co-authored-by: GitHub Action <actions@github.com>
2025-12-01 12:32:14 +05:30
yuri
d146ebfb01 feat(ui): improve misc filtering UX (#13247)
Co-authored-by: Piyush Bhaskar <impiyush0012@gmail.com>
2025-12-01 12:29:49 +05:30
sh
e353399d47 refactor(core): remove usage of unnecessary i18n composable (#13248)
Closes https://github.com/kestra-io/kestra/issues/13200.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-30 21:38:30 +01:00
Miloš Paunović
038083cdf4 fix(core): prevent log lines from breaking too early in firefox (#13243)
Closes https://github.com/kestra-io/kestra/issues/12736.

Related to https://github.com/kestra-io/kestra/pull/13130.
2025-11-28 20:04:19 +01:00
Leonardo Ishida
568e66c75e refactor(core): remove unused component from codebase (#13244)
Closes https://github.com/kestra-io/kestra/issues/13145.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-28 17:56:07 +01:00
Florian Hussonnois
5a8552ad36 refactor(core): remove LogService
Move static methods to Logs utility class
Move purge method to existing ExecutionLogService
2025-11-28 17:48:23 +01:00
Roman Acevedo
da323d792a fix(core): make sure inputs form defaults handle all cases
made with Bart, thx to him
2025-11-28 17:26:11 +01:00
Barthélémy Ledoux
659731813a fix: avoid saving flow.yaml with the rest of namespace files (#13190)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-28 16:30:10 +01:00
Barthélémy Ledoux
b8b20e76ba fix: make Namespace value show a value (#13210)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-28 16:29:47 +01:00
Mahadeva Peruka
cf0b551f8f fix(ui): Executions > outputs is now responsive (#13128)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-28 15:59:05 +01:00
François Delbrayelle
84840fe090 fix(http): safe guard on too big files for Request (#13232) 2025-11-28 13:39:37 +01:00
Roman Acevedo
0dba0367f7 fix(backfills): inputs was always the default one in the ui
- fix https://github.com/kestra-io/kestra/issues/13143
2025-11-28 12:45:54 +01:00
Piyush Bhaskar
905341c185 fix(core): ensure row unselected after actions (#13233) 2025-11-28 16:41:40 +05:30
Loïc Mathieu
b33fbc284d fix(executions): don't ends flowable if any subtasks should be retried
Fixes #11444
2025-11-28 11:00:49 +01:00
yuri
71f1bb9477 feat(ui): improve filters UX (#13192)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
Co-authored-by: Piyush Bhaskar <impiyush0012@gmail.com>
2025-11-28 15:17:20 +05:30
Aditya Pandey
491e286eee refactor(core): remove usage of unnecessary i18n composable (#13229)
Closes https://github.com/kestra-io/kestra/issues/13172.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-28 09:35:05 +01:00
Norman_Osbon
469e230ebd refactor(core): remove usage of unnecessary i18n composable (#13219)
Closes https://github.com/kestra-io/kestra/issues/13203.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-28 08:26:00 +01:00
SungJunBaek
ebb86f6d19 refactor(core): remove usage of unnecessary i18n composable (#13223)
Closes https://github.com/kestra-io/kestra/issues/13202.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-28 08:21:28 +01:00
brian.mulier
b68dcb7bf5 fix(tests): larger timeout for JdbcServiceLivenessCoordinatorTest.shouldReEmitTasksToTheSameWorkerGroup 2025-11-27 19:23:17 +01:00
brian.mulier
65786343ef fix(tests): ensure restart does an assert on RESTARTED state 2025-11-27 19:23:17 +01:00
brian.mulier
d6933b8e49 fix(executions): avoid JdbcExecutor from being stuck due to missing flow
In tests it can occur for example
2025-11-27 19:23:17 +01:00
brian.mulier
8bd5593b2d fix(tests): export path was wrong 2025-11-27 19:23:17 +01:00
brian.mulier
af87713258 fix(tests): attempt to bump amount of executor threads in TestRunner 2025-11-27 19:23:17 +01:00
brian.mulier
371c1281ca fix(tests): SchedulerConditionTest.schedule use Instant instead of ZonedDateTime 2025-11-27 19:23:17 +01:00
brian.mulier
6a111a676c fix(tests): remove @RetryingTest for flaky test 2025-11-27 19:23:17 +01:00
brian.mulier
15da58dbf4 fix(tests): use another db name on webserver to avoid colliding with repositories 2025-11-27 19:23:17 +01:00
brian.mulier
e37e2b0166 fix(tests): better error message on AbstractRunnerTest.concurrencyQueueRestarted 2025-11-27 19:23:17 +01:00
brian.mulier
9f90412237 fix(tests): run WorkingDirectoryTest in same thread due to race condition
related to https://github.com/kestra-io/kestra/issues/13134
2025-11-27 19:23:17 +01:00
brian.mulier
c3d94dc8ff fix(tests): remove JdbcTestUtils.drop usages as it defeats concurrent test runs 2025-11-27 19:23:17 +01:00
brian.mulier
98678deabb fix(tests): use test extension context per classes to follow our concurrent test run strategy 2025-11-27 19:23:17 +01:00
Pratik Dey
248c2154a2 fix: Debug Expression syntax highlighting stops (#13146)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-27 15:54:37 +01:00
YannC
546039e30a fix(webserver): export test with correct path + DateField for execution export (#13197) 2025-11-27 15:43:03 +01:00
Vedant794
27bcb9c347 refactor(core): remove usage of unnecessary i18n composable (#13207)
Closes https://github.com/kestra-io/kestra/issues/13173.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-27 12:58:30 +01:00
Tanbir Ali
3f7b6a0e72 refactor(core): remove usage of unnecessary i18n composable (#13208)
Closes https://github.com/kestra-io/kestra/issues/13204.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-27 12:55:28 +01:00
Mahadeva Peruka
aeca59a3e4 chore(executions): more tweaks to the overview page for small screens (#13206)
Closes https://github.com/kestra-io/kestra/issues/12730.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-27 12:26:49 +01:00
François Delbrayelle
d7caf9ae00 docs: indent properly core plugins examples (#13199) 2025-11-27 11:04:08 +01:00
Miloš Paunović
b5efc27763 fix(core): amend showing of proper date format (#13196)
Related to https://github.com/kestra-io/kestra/pull/13139.

Closes https://github.com/kestra-io/kestra-ee/issues/5969.
2025-11-27 09:05:14 +01:00
Piyush Bhaskar
4909978f7f feat(core): list out the KV and secrets from upstream parent in ns context (#13195) 2025-11-27 13:32:35 +05:30
yuri
f8740871ec chore(core): improve the usability of the logs view (#13191)
- Collapse/Expand functionality shouldn't apply to temporal view
- Specified log level query parameter correctly
2025-11-27 09:00:03 +01:00
RONGALI MOHAN KRISHNA 2400033266
187319ad54 refactor(core): remove usage of unnecessary i18n composable (#13183)
Closes https://github.com/kestra-io/kestra/issues/13175.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-27 08:29:31 +01:00
2400031832
9459d6556b refactor(core): remove usage of unnecessary i18n composable (#13182)
Closes https://github.com/kestra-io/kestra/issues/13176.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-27 08:26:33 +01:00
kkash08
a9d1e9ac4d refactor(core): remove usage of unnecessary i18n composable (#13194)
Closes https://github.com/kestra-io/kestra/issues/13174.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-27 08:22:36 +01:00
Loïc Mathieu
c6c62dbe47 fix(execution): sequential with empty subtasks should ends in SUCCESS
Fixes https://github.com/kestra-io/kestra-ee/issues/5714

It fixes the aforementionned issue as there is a race with Parallel and restart which is caused by subsequent updates on the execution ending in a state where the parallel has no more task to process by didn't ends normally as it should have some.
2025-11-26 18:11:07 +01:00
Loïc Mathieu
8f4bafc666 feat(execution): add an attemps on skipped tasks 2025-11-26 18:11:07 +01:00
Barthélémy Ledoux
e46fbe480e refactor: unify management of anonymous routes (#13181) 2025-11-26 15:56:00 +01:00
YannC
7fd16b24e0 feat: allows to export flows/triggers/executions as a streamed CSV files (#13137) 2025-11-26 14:11:14 +01:00
Richard-Mackey
51529c8ead fix: Add taskId parameter to task log API requests (#13106)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 11:48:49 +01:00
Barthélémy Ledoux
f53135a856 fix: loading icons should work (#13126)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 10:26:45 +01:00
AkhilChowdary2222
bd4eebed32 refactor(core): import toast directly from the composable (#13171)
Closes https://github.com/kestra-io/kestra/issues/12951.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 10:10:10 +01:00
Ravi kumar
f2e7283c72 fix(ui): render level badge and header inline with message (#13130)
Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 14:30:00 +05:30
dependabot[bot]
e31e833ce6 build(deps): bump actions/checkout from 5 to 6
Bumps [actions/checkout](https://github.com/actions/checkout) from 5 to 6.
- [Release notes](https://github.com/actions/checkout/releases)
- [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md)
- [Commits](https://github.com/actions/checkout/compare/v5...v6)

---
updated-dependencies:
- dependency-name: actions/checkout
  dependency-version: '6'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:56:02 +01:00
dependabot[bot]
e7a99bb37f build(deps): bump software.amazon.awssdk.crt:aws-crt
Bumps [software.amazon.awssdk.crt:aws-crt](https://github.com/awslabs/aws-crt-java) from 0.39.4 to 0.40.1.
- [Release notes](https://github.com/awslabs/aws-crt-java/releases)
- [Commits](https://github.com/awslabs/aws-crt-java/compare/v0.39.4...v0.40.1)

---
updated-dependencies:
- dependency-name: software.amazon.awssdk.crt:aws-crt
  dependency-version: 0.40.1
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:55:38 +01:00
dependabot[bot]
1fd4bf7499 build(deps): bump org.sonarqube from 7.0.1.6134 to 7.1.0.6387
Bumps org.sonarqube from 7.0.1.6134 to 7.1.0.6387.

---
updated-dependencies:
- dependency-name: org.sonarqube
  dependency-version: 7.1.0.6387
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:55:15 +01:00
dependabot[bot]
c5851ce254 build(deps): bump com.gorylenko.gradle-git-properties
Bumps com.gorylenko.gradle-git-properties from 2.5.3 to 2.5.4.

---
updated-dependencies:
- dependency-name: com.gorylenko.gradle-git-properties
  dependency-version: 2.5.4
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:54:55 +01:00
dependabot[bot]
1f1976099e build(deps): bump bouncycastleVersion from 1.82 to 1.83
Bumps `bouncycastleVersion` from 1.82 to 1.83.

Updates `org.bouncycastle:bcprov-jdk18on` from 1.82 to 1.83
- [Changelog](https://github.com/bcgit/bc-java/blob/main/docs/releasenotes.html)
- [Commits](https://github.com/bcgit/bc-java/commits)

Updates `org.bouncycastle:bcpg-jdk18on` from 1.82 to 1.83
- [Changelog](https://github.com/bcgit/bc-java/blob/main/docs/releasenotes.html)
- [Commits](https://github.com/bcgit/bc-java/commits)

Updates `org.bouncycastle:bcpkix-jdk18on` from 1.82 to 1.83
- [Changelog](https://github.com/bcgit/bc-java/blob/main/docs/releasenotes.html)
- [Commits](https://github.com/bcgit/bc-java/commits)

---
updated-dependencies:
- dependency-name: org.bouncycastle:bcprov-jdk18on
  dependency-version: '1.83'
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.bouncycastle:bcpg-jdk18on
  dependency-version: '1.83'
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.bouncycastle:bcpkix-jdk18on
  dependency-version: '1.83'
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:54:30 +01:00
dependabot[bot]
6a8e6b414b build(deps): bump flyingSaucerVersion from 10.0.4 to 10.0.5
Bumps `flyingSaucerVersion` from 10.0.4 to 10.0.5.

Updates `org.xhtmlrenderer:flying-saucer-core` from 10.0.4 to 10.0.5
- [Release notes](https://github.com/flyingsaucerproject/flyingsaucer/releases)
- [Changelog](https://github.com/flyingsaucerproject/flyingsaucer/blob/main/CHANGELOG.md)
- [Commits](https://github.com/flyingsaucerproject/flyingsaucer/compare/v10.0.4...v10.0.5)

Updates `org.xhtmlrenderer:flying-saucer-pdf` from 10.0.4 to 10.0.5
- [Release notes](https://github.com/flyingsaucerproject/flyingsaucer/releases)
- [Changelog](https://github.com/flyingsaucerproject/flyingsaucer/blob/main/CHANGELOG.md)
- [Commits](https://github.com/flyingsaucerproject/flyingsaucer/compare/v10.0.4...v10.0.5)

---
updated-dependencies:
- dependency-name: org.xhtmlrenderer:flying-saucer-core
  dependency-version: 10.0.5
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: org.xhtmlrenderer:flying-saucer-pdf
  dependency-version: 10.0.5
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:54:02 +01:00
dependabot[bot]
80d81820c9 build(deps): bump io.pebbletemplates:pebble from 3.2.4 to 4.0.0
Bumps [io.pebbletemplates:pebble](https://github.com/PebbleTemplates/pebble) from 3.2.4 to 4.0.0.
- [Release notes](https://github.com/PebbleTemplates/pebble/releases)
- [Commits](https://github.com/PebbleTemplates/pebble/compare/3.2.4...4.0.0)

---
updated-dependencies:
- dependency-name: io.pebbletemplates:pebble
  dependency-version: 4.0.0
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:53:40 +01:00
dependabot[bot]
0a62957f05 build(deps): bump software.amazon.awssdk:bom from 2.38.9 to 2.39.4
Bumps software.amazon.awssdk:bom from 2.38.9 to 2.39.4.

---
updated-dependencies:
- dependency-name: software.amazon.awssdk:bom
  dependency-version: 2.39.4
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-26 09:52:50 +01:00
dependabot[bot]
2c46bc0c39 build(deps): bump vue-i18n from 11.1.12 to 11.2.2 in /ui (#13167)
Bumps [vue-i18n](https://github.com/intlify/vue-i18n/tree/HEAD/packages/vue-i18n) from 11.1.12 to 11.2.2.
- [Release notes](https://github.com/intlify/vue-i18n/releases)
- [Changelog](https://github.com/intlify/vue-i18n/blob/master/CHANGELOG.md)
- [Commits](https://github.com/intlify/vue-i18n/commits/v11.2.2/packages/vue-i18n)

---
updated-dependencies:
- dependency-name: vue-i18n
  dependency-version: 11.2.2
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-11-26 08:54:26 +01:00
Mahadeva Peruka
f0189c32fc chore(executions): minor tweaks to the overview page for small screens (#13084)
Closes https://github.com/kestra-io/kestra/issues/12730.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 08:47:18 +01:00
zihenzzz
e6058f3d3e chore(executions): minor tweaks to the gantt chart on the execution page for small screens (#13095)
Closes https://github.com/kestra-io/kestra/issues/12735.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 08:41:20 +01:00
Yuvraj Karna
a5ec12c62a refactor(core): import moment as a library directly into the component (#13125)
Closes https://github.com/kestra-io/kestra/issues/12955.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 08:27:28 +01:00
SIVALANAGASHANKARNIVAS
5439d395b1 refactor(core): import moment as a library directly into the component (#13139)
Closes https://github.com/kestra-io/kestra/issues/12956.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-26 08:19:26 +01:00
Piyush Bhaskar
bb363f8832 feat(execution): add execution triggers column (#13158) 2025-11-26 12:46:33 +05:30
Piyush Bhaskar
865aaa1fde refactor(filter): re position refresh and reset (#13142) 2025-11-26 12:46:19 +05:30
Piyush Bhaskar
116e5aad2d fix(core): add nextTick for table rendering (#13148) 2025-11-26 12:39:13 +05:30
Loïc Mathieu
5860ce73bb feat(system): configure Docker Compose termination grace period to 6m
By default, Kestra has a termination grace period of 5m.
To be sure all tasks are terminated, we need to configure Docker Compose with a termination grace period of more than that: 6m.

Part-of: https://github.com/kestra-io/kestra-ee/issues/5556
2025-11-25 17:07:01 +01:00
Nicolas K.
527d80cd74 fix(tests): failing unit test flowProperties (#13138)
Co-authored-by: nKwiatkowski <nkwiatkowski@kestra.io>
2025-11-25 10:29:51 +01:00
Miloš Paunović
c99bd1d4ea fix(core): redirect welcome page action button to flow creation in the enterprise edition (#13136)
Closes https://github.com/kestra-io/kestra-ee/issues/5933.
2025-11-25 08:15:06 +01:00
Hritik Raj
c4a6ea617f refactor(ui): migrate Toc.vue to Vue 3 Composition API with full Type… (#13113) 2025-11-24 16:05:33 +05:30
Pratik Dey
a4b0beaf63 refactor(core): remove unused component from codebase (#13123)
Closes https://github.com/kestra-io/kestra/issues/12960.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-24 09:49:16 +01:00
Raúl Soto
a5847aeb3a refactor(core): remove unused component from codebase (#13111)
Closes https://github.com/kestra-io/kestra/issues/12961.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-24 08:00:32 +01:00
Ronin@73
49bbc15d91 refactor(core): remove unused component from codebase (#13112)
Closes https://github.com/kestra-io/kestra/issues/12959.

Co-authored-by: AtulRaghuvanshi73 <atul.raghuvanshi73@gmail.com>
Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-24 07:57:05 +01:00
Loïc Mathieu
9d6694f807 fix(system): WorkerTask should not FAILED when interrupting so they would be resubmitted
When a Worker is stopping, it will first wait for all running tasks to stop, then kill them. For those that didn't implement kill their thread would be interrupted.

But if the task is properly killed, or support interrupts (like the Sleep task), it would ends in FAILED then a WorkerTaskWould be send that would fail the flow preventing the WorkerTask to be resubmitted.

We nows check if the worker is terminating and should resubmit, in this case we didn't emit any WorkerTaskResult

Fixes #13108
Part-of: https://github.com/kestra-io/kestra-ee/issues/5556
2025-11-21 16:58:36 +01:00
Florian Hussonnois
eb51c5be37 feat(flows): add new check conditions
Adds new property 'checks' on flow in order to allow
pre-conditions to be evaluated before execution

Fixes: kestra-io/kestra-ee#5759
2025-11-21 14:25:01 +01:00
Arghyadeep
90ee720d49 refactor(core): remove usage of unnecessary i18n composable (#13103)
Closes https://github.com/kestra-io/kestra/issues/12949.

Co-authored-by: Miloš Paunović <paun992@hotmail.com>
2025-11-21 08:13:09 +01:00
vishnuvarthan
fd259082a6 refactor(core): remove usage of unnecessary i18n composable (#13104)
Closes https://github.com/kestra-io/kestra/issues/12970.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-21 08:02:30 +01:00
François Delbrayelle
b5323f969c chore: update core plugins icons with new look & feel (#13102) 2025-11-20 16:56:09 +01:00
Barthélémy Ledoux
6c826e93c8 feat: implement unsaved changes handling with dedicated store and dialog (#13090) 2025-11-20 15:47:25 +01:00
Georg Traar
aae3e6605d docs(foreach): improve explanation of concurrencyLimit semantics (#13085)
* docs(foreach): improve explanation of concurrencyLimit semantics

---------

Co-authored-by: Georg Traar <georg@crate.io>
2025-11-20 13:08:45 +01:00
Barthélémy Ledoux
ea17077b0a tests(e2e): reproduce and fix flake (#13075) 2025-11-20 11:59:00 +01:00
Barthélémy Ledoux
117200eaab fix: add defaultScope and defaultTimeRange props to various components (#13097) 2025-11-20 11:56:48 +01:00
YannC
3216611828 fix: make sure variables from ExecutionTrigger has AdditionalPropertiesValue to true (#13096) 2025-11-20 11:35:39 +01:00
Barthélémy Ledoux
1173eb2dde fix(filters): make restoreUrl work better
closes #13082

fix redirection scheme:
remove timeout for restoreUrl
remove use of the restoreUrl hook in filters since it is only saved in useDataTableAction
to keep the order intact (restore first, default second) we add 2 nextTick() in useDefaultFilter() one for router the other for restoreUrl.
2025-11-20 11:03:02 +01:00
Pradumna Saraf
360b58a851 fix: failing sanity checks 2025-11-20 11:00:52 +01:00
Loïc Mathieu
57e288abdd fix(execution): improve property skip cache
When using Property.ofExpression(), the cache should never be used as this is usually used as providing a default value inside a task, which can change from rendering to rendering as it's an expression.

Also retain skipCache in a boolean so it can be rendered more than 2 times ans still skip the cache.

It should prevent future issues like #13027
2025-11-20 10:23:37 +01:00
YannC
7fa14eb3f5 chore(API): apiResponse annotation for type return (#13088) 2025-11-20 09:47:32 +01:00
Adriana Arroyo Fernandez
0ed2b0a53c refactor(core): remove usage of unnecessary i18n composable (#13083)
Closes https://github.com/kestra-io/kestra/issues/12950.

Co-authored-by: MilosPaunovic <paun992@hotmail.com>
2025-11-20 07:58:18 +01:00
YannC
68ace7a59b fix: loop until avoid taskrun duplication + handle submitted status (#13023)
* fix: in LoopUntil, missing handle of submitted task in the resolveWaitForNext method

* fix: remove useless code that was generating duplicate taskrun
2025-11-19 15:14:18 +01:00
Nicolas K.
105b1b36e5 feat(flows): improve human task in bulk method, kill and force run (#13067)
Co-authored-by: nKwiatkowski <nkwiatkowski@kestra.io>
2025-11-19 14:53:59 +01:00
Iulian Ghita
15e82f65c6 fix(ui): make plugins pages responsive (#12829)
Co-authored-by: Piyush Bhaskar <102078527+Piyush-r-bhaskar@users.noreply.github.com>
2025-11-19 18:31:15 +05:30
dependabot[bot]
aec75bb673 build(deps): bump nl.basjes.gitignore:gitignore-reader
Bumps [nl.basjes.gitignore:gitignore-reader](https://github.com/nielsbasjes/codeowners) from 1.12.2 to 1.13.0.
- [Release notes](https://github.com/nielsbasjes/codeowners/releases)
- [Changelog](https://github.com/nielsbasjes/codeowners/blob/main/CHANGELOG.md)
- [Commits](https://github.com/nielsbasjes/codeowners/compare/v1.12.2...v1.13.0)

---
updated-dependencies:
- dependency-name: nl.basjes.gitignore:gitignore-reader
  dependency-version: 1.13.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 13:40:34 +01:00
Florian Hussonnois
f489678532 fix(trigger): TimeBetween should support cross-midnight range 2025-11-19 13:32:14 +01:00
Florian Hussonnois
79fc5a3f24 fix(trigger): fix trigger evaluation with TimeBetween condition (#10900)
Fixes: #10900
2025-11-19 13:32:14 +01:00
Piyush Bhaskar
312ec2c36b chore(core): remove import warning (#13071) 2025-11-19 17:27:00 +05:30
Piyush Bhaskar
d57150e69c chore(test): fix frontend failing test (#13070) 2025-11-19 17:23:16 +05:30
Miloš Paunović
4b25232d4e build(core): remove unused dependency (#13064) 2025-11-19 12:33:19 +01:00
Loïc Mathieu
1d1a065833 fix(tests): lower termination grace period
If a test mis-behave, for ex starting an execution but not terminating it, as the default termination grace period is 5mn it can take very long time to wait for post-test terminaison.
Switching to a termination grace period of 5s may help.

I also detect that the ExecutionControllerRunner test when launching the test suite, would not properly kill the `sleep-long` flow so waiting for it to complete, or the termination grace period. When a test that use this flow is launched separatly it works properly. As a safety net I reduce the sleep from 5mn to 30s.
2025-11-19 11:58:06 +01:00
dependabot[bot]
d6ecbadee1 build(deps): bump kafkaVersion from 4.1.0 to 4.1.1
Bumps `kafkaVersion` from 4.1.0 to 4.1.1.

Updates `org.apache.kafka:kafka-clients` from 4.1.0 to 4.1.1

Updates `org.apache.kafka:kafka-streams` from 4.1.0 to 4.1.1

Updates `org.apache.kafka:kafka-streams-test-utils` from 4.1.0 to 4.1.1

---
updated-dependencies:
- dependency-name: org.apache.kafka:kafka-clients
  dependency-version: 4.1.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: org.apache.kafka:kafka-streams
  dependency-version: 4.1.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: org.apache.kafka:kafka-streams-test-utils
  dependency-version: 4.1.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:51:01 +01:00
dependabot[bot]
205605060d build(deps): bump com.google.cloud:libraries-bom from 26.71.0 to 26.72.0
Bumps [com.google.cloud:libraries-bom](https://github.com/googleapis/java-cloud-bom) from 26.71.0 to 26.72.0.
- [Release notes](https://github.com/googleapis/java-cloud-bom/releases)
- [Changelog](https://github.com/googleapis/java-cloud-bom/blob/main/release-please-config.json)
- [Commits](https://github.com/googleapis/java-cloud-bom/compare/v26.71.0...v26.72.0)

---
updated-dependencies:
- dependency-name: com.google.cloud:libraries-bom
  dependency-version: 26.72.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:50:38 +01:00
dependabot[bot]
6ef3a00e16 build(deps): bump peter-evans/repository-dispatch from 4.0.0 to 4.0.1
Bumps [peter-evans/repository-dispatch](https://github.com/peter-evans/repository-dispatch) from 4.0.0 to 4.0.1.
- [Release notes](https://github.com/peter-evans/repository-dispatch/releases)
- [Commits](5fc4efd1a4...28959ce8df)

---
updated-dependencies:
- dependency-name: peter-evans/repository-dispatch
  dependency-version: 4.0.1
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:48:57 +01:00
dependabot[bot]
f70d612878 build(deps): bump actions/github-script from 7 to 8
Bumps [actions/github-script](https://github.com/actions/github-script) from 7 to 8.
- [Release notes](https://github.com/actions/github-script/releases)
- [Commits](https://github.com/actions/github-script/compare/v7...v8)

---
updated-dependencies:
- dependency-name: actions/github-script
  dependency-version: '8'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:48:35 +01:00
dependabot[bot]
0b345c03d1 build(deps): bump net.e175.klaus:zip-prefixer from 0.3.1 to 0.4.0
Bumps [net.e175.klaus:zip-prefixer](https://github.com/klausbrunner/zip-prefixer) from 0.3.1 to 0.4.0.
- [Commits](https://github.com/klausbrunner/zip-prefixer/commits)

---
updated-dependencies:
- dependency-name: net.e175.klaus:zip-prefixer
  dependency-version: 0.4.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:48:03 +01:00
dependabot[bot]
ecb508f797 build(deps): bump io.qameta.allure:allure-bom from 2.30.0 to 2.31.0
Bumps [io.qameta.allure:allure-bom](https://github.com/allure-framework/allure-java) from 2.30.0 to 2.31.0.
- [Release notes](https://github.com/allure-framework/allure-java/releases)
- [Commits](https://github.com/allure-framework/allure-java/compare/2.30.0...2.31.0)

---
updated-dependencies:
- dependency-name: io.qameta.allure:allure-bom
  dependency-version: 2.31.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:47:43 +01:00
dependabot[bot]
38caea2568 build(deps): bump flyingSaucerVersion from 10.0.3 to 10.0.4
Bumps `flyingSaucerVersion` from 10.0.3 to 10.0.4.

Updates `org.xhtmlrenderer:flying-saucer-core` from 10.0.3 to 10.0.4
- [Release notes](https://github.com/flyingsaucerproject/flyingsaucer/releases)
- [Changelog](https://github.com/flyingsaucerproject/flyingsaucer/blob/main/CHANGELOG.md)
- [Commits](https://github.com/flyingsaucerproject/flyingsaucer/compare/v10.0.3...v10.0.4)

Updates `org.xhtmlrenderer:flying-saucer-pdf` from 10.0.3 to 10.0.4
- [Release notes](https://github.com/flyingsaucerproject/flyingsaucer/releases)
- [Changelog](https://github.com/flyingsaucerproject/flyingsaucer/blob/main/CHANGELOG.md)
- [Commits](https://github.com/flyingsaucerproject/flyingsaucer/compare/v10.0.3...v10.0.4)

---
updated-dependencies:
- dependency-name: org.xhtmlrenderer:flying-saucer-core
  dependency-version: 10.0.4
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: org.xhtmlrenderer:flying-saucer-pdf
  dependency-version: 10.0.4
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:47:22 +01:00
dependabot[bot]
cdad732576 build(deps): bump org.wiremock:wiremock-jetty12 from 3.13.1 to 3.13.2
Bumps [org.wiremock:wiremock-jetty12](https://github.com/wiremock/wiremock) from 3.13.1 to 3.13.2.
- [Release notes](https://github.com/wiremock/wiremock/releases)
- [Commits](https://github.com/wiremock/wiremock/compare/3.13.1...3.13.2)

---
updated-dependencies:
- dependency-name: org.wiremock:wiremock-jetty12
  dependency-version: 3.13.2
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:46:15 +01:00
dependabot[bot]
0dd4cb963f build(deps): bump org.apache.commons:commons-lang3 from 3.19.0 to 3.20.0
Bumps org.apache.commons:commons-lang3 from 3.19.0 to 3.20.0.

---
updated-dependencies:
- dependency-name: org.apache.commons:commons-lang3
  dependency-version: 3.20.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:45:46 +01:00
dependabot[bot]
c35ca82356 build(deps): bump software.amazon.awssdk:bom from 2.38.4 to 2.38.9
Bumps software.amazon.awssdk:bom from 2.38.4 to 2.38.9.

---
updated-dependencies:
- dependency-name: software.amazon.awssdk:bom
  dependency-version: 2.38.9
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
2025-11-19 11:45:21 +01:00
Miloš Paunović
db6cb93df4 chore(deps): regular dependency update (#13065)
Performing a weekly round of dependency updates in the NPM ecosystem to keep everything up to date.
2025-11-19 09:45:28 +01:00
YannC
1f8d2ea918 fix: use pull request head sha instead of commit.sha (#13063) 2025-11-19 09:34:34 +01:00
dependabot[bot]
3c09a38eed build(deps): bump the patch group in /ui with 6 updates (#13057)
Bumps the patch group in /ui with 6 updates:

| Package | From | To |
| --- | --- | --- |
| [el-table-infinite-scroll](https://github.com/yujinpan/el-table-infinite-scroll) | `3.0.7` | `3.0.8` |
| [element-plus](https://github.com/element-plus/element-plus) | `2.11.7` | `2.11.8` |
| [@vitejs/plugin-vue](https://github.com/vitejs/vite-plugin-vue/tree/HEAD/packages/plugin-vue) | `6.0.1` | `6.0.2` |
| [@vitejs/plugin-vue-jsx](https://github.com/vitejs/vite-plugin-vue/tree/HEAD/packages/plugin-vue-jsx) | `5.1.1` | `5.1.2` |
| [rolldown-vite](https://github.com/vitejs/rolldown-vite/tree/HEAD/packages/vite) | `7.2.5` | `7.2.6` |
| [vue-tsc](https://github.com/vuejs/language-tools/tree/HEAD/packages/tsc) | `3.1.3` | `3.1.4` |


Updates `el-table-infinite-scroll` from 3.0.7 to 3.0.8
- [Changelog](https://github.com/yujinpan/el-table-infinite-scroll/blob/v3/CHANGELOG.md)
- [Commits](https://github.com/yujinpan/el-table-infinite-scroll/compare/v3.0.7...v3.0.8)

Updates `element-plus` from 2.11.7 to 2.11.8
- [Release notes](https://github.com/element-plus/element-plus/releases)
- [Changelog](https://github.com/element-plus/element-plus/blob/dev/CHANGELOG.en-US.md)
- [Commits](https://github.com/element-plus/element-plus/compare/2.11.7...2.11.8)

Updates `@vitejs/plugin-vue` from 6.0.1 to 6.0.2
- [Release notes](https://github.com/vitejs/vite-plugin-vue/releases)
- [Changelog](https://github.com/vitejs/vite-plugin-vue/blob/main/packages/plugin-vue/CHANGELOG.md)
- [Commits](https://github.com/vitejs/vite-plugin-vue/commits/plugin-vue@6.0.2/packages/plugin-vue)

Updates `@vitejs/plugin-vue-jsx` from 5.1.1 to 5.1.2
- [Release notes](https://github.com/vitejs/vite-plugin-vue/releases)
- [Changelog](https://github.com/vitejs/vite-plugin-vue/blob/main/packages/plugin-vue-jsx/CHANGELOG.md)
- [Commits](https://github.com/vitejs/vite-plugin-vue/commits/plugin-vue@5.1.2/packages/plugin-vue-jsx)

Updates `rolldown-vite` from 7.2.5 to 7.2.6
- [Release notes](https://github.com/vitejs/rolldown-vite/releases)
- [Changelog](https://github.com/vitejs/rolldown-vite/blob/rolldown-vite/packages/vite/CHANGELOG.md)
- [Commits](https://github.com/vitejs/rolldown-vite/commits/v7.2.6/packages/vite)

Updates `vue-tsc` from 3.1.3 to 3.1.4
- [Release notes](https://github.com/vuejs/language-tools/releases)
- [Changelog](https://github.com/vuejs/language-tools/blob/master/CHANGELOG.md)
- [Commits](https://github.com/vuejs/language-tools/commits/v3.1.4/packages/tsc)

---
updated-dependencies:
- dependency-name: el-table-infinite-scroll
  dependency-version: 3.0.8
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: patch
- dependency-name: element-plus
  dependency-version: 2.11.8
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: patch
- dependency-name: "@vitejs/plugin-vue"
  dependency-version: 6.0.2
  dependency-type: direct:development
  update-type: version-update:semver-patch
  dependency-group: patch
- dependency-name: "@vitejs/plugin-vue-jsx"
  dependency-version: 5.1.2
  dependency-type: direct:development
  update-type: version-update:semver-patch
  dependency-group: patch
- dependency-name: rolldown-vite
  dependency-version: 7.2.6
  dependency-type: direct:development
  update-type: version-update:semver-patch
  dependency-group: patch
- dependency-name: vue-tsc
  dependency-version: 3.1.4
  dependency-type: direct:development
  update-type: version-update:semver-patch
  dependency-group: patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-11-19 08:54:11 +01:00
dependabot[bot]
0525e3ece6 build(deps): bump the build group in /ui with 6 updates (#13044)
Bumps the build group in /ui with 6 updates:

| Package | From | To |
| --- | --- | --- |
| [@rollup/rollup-darwin-arm64](https://github.com/rollup/rollup) | `4.53.2` | `4.53.3` |
| [@rollup/rollup-darwin-x64](https://github.com/rollup/rollup) | `4.53.2` | `4.53.3` |
| [@rollup/rollup-linux-x64-gnu](https://github.com/rollup/rollup) | `4.53.2` | `4.53.3` |
| [@swc/core-darwin-arm64](https://github.com/swc-project/swc) | `1.15.1` | `1.15.2` |
| [@swc/core-darwin-x64](https://github.com/swc-project/swc) | `1.15.1` | `1.15.2` |
| [@swc/core-linux-x64-gnu](https://github.com/swc-project/swc) | `1.15.1` | `1.15.2` |


Updates `@rollup/rollup-darwin-arm64` from 4.53.2 to 4.53.3
- [Release notes](https://github.com/rollup/rollup/releases)
- [Changelog](https://github.com/rollup/rollup/blob/master/CHANGELOG.md)
- [Commits](https://github.com/rollup/rollup/compare/v4.53.2...v4.53.3)

Updates `@rollup/rollup-darwin-x64` from 4.53.2 to 4.53.3
- [Release notes](https://github.com/rollup/rollup/releases)
- [Changelog](https://github.com/rollup/rollup/blob/master/CHANGELOG.md)
- [Commits](https://github.com/rollup/rollup/compare/v4.53.2...v4.53.3)

Updates `@rollup/rollup-linux-x64-gnu` from 4.53.2 to 4.53.3
- [Release notes](https://github.com/rollup/rollup/releases)
- [Changelog](https://github.com/rollup/rollup/blob/master/CHANGELOG.md)
- [Commits](https://github.com/rollup/rollup/compare/v4.53.2...v4.53.3)

Updates `@swc/core-darwin-arm64` from 1.15.1 to 1.15.2
- [Release notes](https://github.com/swc-project/swc/releases)
- [Changelog](https://github.com/swc-project/swc/blob/main/CHANGELOG.md)
- [Commits](https://github.com/swc-project/swc/compare/v1.15.1...v1.15.2)

Updates `@swc/core-darwin-x64` from 1.15.1 to 1.15.2
- [Release notes](https://github.com/swc-project/swc/releases)
- [Changelog](https://github.com/swc-project/swc/blob/main/CHANGELOG.md)
- [Commits](https://github.com/swc-project/swc/compare/v1.15.1...v1.15.2)

Updates `@swc/core-linux-x64-gnu` from 1.15.1 to 1.15.2
- [Release notes](https://github.com/swc-project/swc/releases)
- [Changelog](https://github.com/swc-project/swc/blob/main/CHANGELOG.md)
- [Commits](https://github.com/swc-project/swc/compare/v1.15.1...v1.15.2)

---
updated-dependencies:
- dependency-name: "@rollup/rollup-darwin-arm64"
  dependency-version: 4.53.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@rollup/rollup-darwin-x64"
  dependency-version: 4.53.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@rollup/rollup-linux-x64-gnu"
  dependency-version: 4.53.3
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@swc/core-darwin-arm64"
  dependency-version: 1.15.2
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@swc/core-darwin-x64"
  dependency-version: 1.15.2
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
- dependency-name: "@swc/core-linux-x64-gnu"
  dependency-version: 1.15.2
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: build
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-11-19 08:31:51 +01:00
587 changed files with 18711 additions and 9242 deletions

View File

@@ -2,6 +2,7 @@ name: Bug report
description: Report a bug or unexpected behavior in the project
labels: ["bug", "area/backend", "area/frontend"]
type: Bug
body:
- type: markdown

View File

@@ -2,6 +2,7 @@ name: Feature request
description: Suggest a new feature or improvement to enhance the project
labels: ["enhancement", "area/backend", "area/frontend"]
type: Feature
body:
- type: textarea

View File

@@ -26,7 +26,7 @@ updates:
open-pull-requests-limit: 50
labels: ["dependency-upgrade", "area/backend"]
ignore:
# Ignore versions of Protobuf that are equal to or greater than 4.0.0 as Orc still uses 3
# Ignore versions of Protobuf >= 4.0.0 because Orc still uses version 3
- dependency-name: "com.google.protobuf:*"
versions: ["[4,)"]
@@ -44,68 +44,73 @@ updates:
build:
applies-to: version-updates
patterns: ["@esbuild/*", "@rollup/*", "@swc/*"]
types:
applies-to: version-updates
patterns: ["@types/*"]
storybook:
applies-to: version-updates
patterns: ["@storybook/*"]
patterns: ["storybook*", "@storybook/*"]
vitest:
applies-to: version-updates
patterns: ["vitest", "@vitest/*"]
patch:
major:
update-types: ["major"]
applies-to: version-updates
exclude-patterns: [
"@esbuild/*",
"@rollup/*",
"@swc/*",
"@types/*",
"storybook*",
"@storybook/*",
"vitest",
"@vitest/*",
# Temporary exclusion of these packages from major updates
"eslint-plugin-storybook",
"eslint-plugin-vue",
]
minor:
update-types: ["minor"]
applies-to: version-updates
exclude-patterns: [
"@esbuild/*",
"@rollup/*",
"@swc/*",
"@types/*",
"storybook*",
"@storybook/*",
"vitest",
"@vitest/*",
# Temporary exclusion of these packages from minor updates
"moment-timezone",
"monaco-editor",
]
patch:
update-types: ["patch"]
applies-to: version-updates
patterns: ["*"]
exclude-patterns:
[
"@esbuild/*",
"@rollup/*",
"@swc/*",
"@types/*",
"storybook*",
"@storybook/*",
"vitest",
"@vitest/*",
]
update-types: ["patch"]
minor:
applies-to: version-updates
patterns: ["*"]
exclude-patterns: [
"@esbuild/*",
"@rollup/*",
"@swc/*",
"@types/*",
"@storybook/*",
"vitest",
"@vitest/*",
# Temporary exclusion of packages below from minor updates
"moment-timezone",
"monaco-editor",
]
update-types: ["minor"]
major:
applies-to: version-updates
patterns: ["*"]
exclude-patterns: [
"@esbuild/*",
"@rollup/*",
"@swc/*",
"@types/*",
"@storybook/*",
"vitest",
"@vitest/*",
# Temporary exclusion of packages below from major updates
"eslint-plugin-storybook",
"eslint-plugin-vue",
]
update-types: ["major"]
ignore:
# Ignore updates to monaco-yaml, version is pinned to 5.3.1 due to patch-package script additions
- dependency-name: "monaco-yaml"
versions:
- ">=5.3.2"
# Ignore updates of version 1.x, as we're using the beta of 2.x (still in beta)
ignore:
# Ignore updates to monaco-yaml; version is pinned to 5.3.1 due to patch-package script additions
- dependency-name: "monaco-yaml"
versions: [">=5.3.2"]
# Ignore updates of version 1.x for vue-virtual-scroller, as the project uses the beta of 2.x
- dependency-name: "vue-virtual-scroller"
versions:
- "1.x"
versions: ["1.x"]

View File

@@ -20,7 +20,7 @@ jobs:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v6
name: Checkout
with:
fetch-depth: 0

View File

@@ -27,7 +27,7 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v5
uses: actions/checkout@v6
with:
# We must fetch at least the immediate parents so that if this is
# a pull request then we can checkout the head.

View File

@@ -33,7 +33,7 @@ jobs:
exit 1;
fi
# Checkout
- uses: actions/checkout@v5
- uses: actions/checkout@v6
with:
fetch-depth: 0
path: kestra

View File

@@ -39,7 +39,7 @@ jobs:
# Checkout
- name: Checkout
uses: actions/checkout@v5
uses: actions/checkout@v6
with:
fetch-depth: 0
token: ${{ secrets.GH_PERSONAL_TOKEN }}

View File

@@ -28,7 +28,7 @@ jobs:
steps:
# Targeting develop branch from develop
- name: Trigger EE Workflow (develop push, no payload)
uses: peter-evans/repository-dispatch@5fc4efd1a4797ddb68ffd0714a238564e4cc0e6f
uses: peter-evans/repository-dispatch@28959ce8df70de7be546dd1250a005dd32156697
if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' }}
with:
token: ${{ secrets.GH_PERSONAL_TOKEN }}
@@ -64,6 +64,7 @@ jobs:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
publish-develop-maven:

View File

@@ -16,7 +16,7 @@ jobs:
- name: Check EE repo for branch with same name
if: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork == false }}
id: check-ee-branch
uses: actions/github-script@v7
uses: actions/github-script@v8
with:
github-token: ${{ secrets.GH_PERSONAL_TOKEN }}
script: |
@@ -40,7 +40,7 @@ jobs:
# Targeting pull request (only if not from a fork and EE has no branch with same name)
- name: Trigger EE Workflow (pull request, with payload)
uses: peter-evans/repository-dispatch@5fc4efd1a4797ddb68ffd0714a238564e4cc0e6f
uses: peter-evans/repository-dispatch@28959ce8df70de7be546dd1250a005dd32156697
if: ${{ github.event_name == 'pull_request'
&& github.event.pull_request.number != ''
&& github.event.pull_request.head.repo.fork == false
@@ -50,7 +50,7 @@ jobs:
repository: kestra-io/kestra-ee
event-type: "oss-updated"
client-payload: >-
{"commit_sha":"${{ github.sha }}","pr_repo":"${{ github.repository }}"}
{"commit_sha":"${{ github.event.pull_request.head.sha }}","pr_repo":"${{ github.repository }}"}
file-changes:
if: ${{ github.event.pull_request.draft == false }}

View File

@@ -32,3 +32,4 @@ jobs:
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}

View File

@@ -17,7 +17,7 @@ jobs:
runs-on: ubuntu-latest
steps:
# Checkout
- uses: actions/checkout@v5
- uses: actions/checkout@v6
with:
fetch-depth: 0
@@ -58,7 +58,7 @@ jobs:
actions: read
steps:
# Checkout
- uses: actions/checkout@v5
- uses: actions/checkout@v6
with:
fetch-depth: 0
@@ -95,7 +95,7 @@ jobs:
actions: read
steps:
# Checkout
- uses: actions/checkout@v5
- uses: actions/checkout@v6
with:
fetch-depth: 0

View File

@@ -7,7 +7,7 @@ buildscript {
}
dependencies {
classpath "net.e175.klaus:zip-prefixer:0.3.1"
classpath "net.e175.klaus:zip-prefixer:0.4.0"
}
}
@@ -21,7 +21,7 @@ plugins {
// test
id "com.adarshr.test-logger" version "4.0.0"
id "org.sonarqube" version "7.0.1.6134"
id "org.sonarqube" version "7.2.0.6526"
id 'jacoco-report-aggregation'
// helper
@@ -32,7 +32,7 @@ plugins {
// release
id 'net.researchgate.release' version '3.1.0'
id "com.gorylenko.gradle-git-properties" version "2.5.3"
id "com.gorylenko.gradle-git-properties" version "2.5.4"
id 'signing'
id "com.vanniktech.maven.publish" version "0.35.0"
@@ -223,13 +223,13 @@ subprojects {subProj ->
t.environment 'ENV_TEST2', "Pass by env"
if (subProj.name == 'core' || subProj.name == 'jdbc-h2' || subProj.name == 'jdbc-mysql' || subProj.name == 'jdbc-postgres') {
// JUnit 5 parallel settings
t.systemProperty 'junit.jupiter.execution.parallel.enabled', 'true'
t.systemProperty 'junit.jupiter.execution.parallel.mode.default', 'concurrent'
t.systemProperty 'junit.jupiter.execution.parallel.mode.classes.default', 'same_thread'
t.systemProperty 'junit.jupiter.execution.parallel.config.strategy', 'dynamic'
}
// if (subProj.name == 'core' || subProj.name == 'jdbc-h2' || subProj.name == 'jdbc-mysql' || subProj.name == 'jdbc-postgres') {
// // JUnit 5 parallel settings
// t.systemProperty 'junit.jupiter.execution.parallel.enabled', 'true'
// t.systemProperty 'junit.jupiter.execution.parallel.mode.default', 'concurrent'
// t.systemProperty 'junit.jupiter.execution.parallel.mode.classes.default', 'same_thread'
// t.systemProperty 'junit.jupiter.execution.parallel.config.strategy', 'dynamic'
// }
}
tasks.register('flakyTest', Test) { Test t ->

View File

@@ -93,7 +93,7 @@ public class App implements Callable<Integer> {
try {
exitCode = new CommandLine(cls, new MicronautFactory(applicationContext)).execute(args);
} catch (CommandLine.InitializationException e){
System.err.println("Could not initialize picoli ComandLine, err: " + e.getMessage());
System.err.println("Could not initialize picocli CommandLine, err: " + e.getMessage());
e.printStackTrace();
exitCode = 1;
}

View File

@@ -10,7 +10,8 @@ import picocli.CommandLine;
description = "populate metadata for entities",
subcommands = {
KvMetadataMigrationCommand.class,
SecretsMetadataMigrationCommand.class
SecretsMetadataMigrationCommand.class,
NsFilesMetadataMigrationCommand.class
}
)
@Slf4j

View File

@@ -1,47 +1,51 @@
package io.kestra.cli.commands.migrations.metadata;
import com.google.common.annotations.VisibleForTesting;
import io.kestra.core.models.kv.PersistedKvMetadata;
import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
import io.kestra.core.repositories.FlowRepositoryInterface;
import io.kestra.core.repositories.KvMetadataRepositoryInterface;
import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
import io.kestra.core.storages.FileAttributes;
import io.kestra.core.storages.StorageContext;
import io.kestra.core.storages.StorageInterface;
import io.kestra.core.storages.kv.InternalKVStore;
import io.kestra.core.storages.kv.KVEntry;
import io.kestra.core.tenant.TenantService;
import jakarta.inject.Inject;
import io.kestra.core.utils.NamespaceUtils;
import jakarta.inject.Singleton;
import lombok.AllArgsConstructor;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
import java.nio.file.NoSuchFileException;
import java.time.Instant;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static io.kestra.core.utils.Rethrow.throwConsumer;
import static io.kestra.core.utils.Rethrow.throwFunction;
@Singleton
@AllArgsConstructor
public class MetadataMigrationService {
@Inject
private TenantService tenantService;
protected FlowRepositoryInterface flowRepository;
protected TenantService tenantService;
protected KvMetadataRepositoryInterface kvMetadataRepository;
protected NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository;
protected StorageInterface storageInterface;
protected NamespaceUtils namespaceUtils;
@Inject
private FlowRepositoryInterface flowRepository;
@Inject
private KvMetadataRepositoryInterface kvMetadataRepository;
@Inject
private StorageInterface storageInterface;
protected Map<String, List<String>> namespacesPerTenant() {
@VisibleForTesting
public Map<String, List<String>> namespacesPerTenant() {
String tenantId = tenantService.resolveTenant();
return Map.of(tenantId, flowRepository.findDistinctNamespace(tenantId));
return Map.of(tenantId, Stream.concat(
Stream.of(namespaceUtils.getSystemFlowNamespace()),
flowRepository.findDistinctNamespace(tenantId).stream()
).map(NamespaceUtils::asTree).flatMap(Collection::stream).distinct().toList());
}
public void kvMigration() throws IOException {
@@ -49,7 +53,9 @@ public class MetadataMigrationService {
.flatMap(namespacesForTenant -> namespacesForTenant.getValue().stream().map(namespace -> Map.entry(namespacesForTenant.getKey(), namespace)))
.flatMap(throwFunction(namespaceForTenant -> {
InternalKVStore kvStore = new InternalKVStore(namespaceForTenant.getKey(), namespaceForTenant.getValue(), storageInterface, kvMetadataRepository);
List<FileAttributes> list = listAllFromStorage(storageInterface, namespaceForTenant.getKey(), namespaceForTenant.getValue());
List<FileAttributes> list = listAllFromStorage(storageInterface, StorageContext::kvPrefix, namespaceForTenant.getKey(), namespaceForTenant.getValue()).stream()
.map(PathAndAttributes::attributes)
.toList();
Map<Boolean, List<KVEntry>> entriesByIsExpired = list.stream()
.map(throwFunction(fileAttributes -> KVEntry.from(namespaceForTenant.getValue(), fileAttributes)))
.collect(Collectors.partitioningBy(kvEntry -> Optional.ofNullable(kvEntry.expirationDate()).map(expirationDate -> Instant.now().isAfter(expirationDate)).orElse(false)));
@@ -75,15 +81,39 @@ public class MetadataMigrationService {
}));
}
public void nsFilesMigration() throws IOException {
this.namespacesPerTenant().entrySet().stream()
.flatMap(namespacesForTenant -> namespacesForTenant.getValue().stream().map(namespace -> Map.entry(namespacesForTenant.getKey(), namespace)))
.flatMap(throwFunction(namespaceForTenant -> {
List<PathAndAttributes> list = listAllFromStorage(storageInterface, StorageContext::namespaceFilePrefix, namespaceForTenant.getKey(), namespaceForTenant.getValue());
return list.stream()
.map(pathAndAttributes -> NamespaceFileMetadata.of(namespaceForTenant.getKey(), namespaceForTenant.getValue(), pathAndAttributes.path(), pathAndAttributes.attributes()));
}))
.forEach(throwConsumer(nsFileMetadata -> {
if (namespaceFileMetadataRepository.findByPath(nsFileMetadata.getTenantId(), nsFileMetadata.getNamespace(), nsFileMetadata.getPath()).isEmpty()) {
namespaceFileMetadataRepository.save(nsFileMetadata);
}
}));
}
public void secretMigration() throws Exception {
throw new UnsupportedOperationException("Secret migration is not needed in the OSS version");
}
private static List<FileAttributes> listAllFromStorage(StorageInterface storage, String tenant, String namespace) throws IOException {
private static List<PathAndAttributes> listAllFromStorage(StorageInterface storage, Function<String, String> prefixFunction, String tenant, String namespace) throws IOException {
try {
return storage.list(tenant, namespace, URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.kvPrefix(namespace)));
} catch (FileNotFoundException e) {
String prefix = prefixFunction.apply(namespace);
if (!storage.exists(tenant, namespace, URI.create(StorageContext.KESTRA_PROTOCOL + prefix))) {
return Collections.emptyList();
}
return storage.allByPrefix(tenant, namespace, URI.create(StorageContext.KESTRA_PROTOCOL + prefix + "/"), true).stream()
.map(throwFunction(uri -> new PathAndAttributes(uri.getPath().substring(prefix.length()), storage.getAttributes(tenant, namespace, uri))))
.toList();
} catch (FileNotFoundException | NoSuchFileException e) {
return Collections.emptyList();
}
}
public record PathAndAttributes(String path, FileAttributes attributes) {}
}

View File

@@ -0,0 +1,31 @@
package io.kestra.cli.commands.migrations.metadata;
import io.kestra.cli.AbstractCommand;
import jakarta.inject.Inject;
import jakarta.inject.Provider;
import lombok.extern.slf4j.Slf4j;
import picocli.CommandLine;
@CommandLine.Command(
name = "nsfiles",
description = "populate metadata for Namespace Files"
)
@Slf4j
public class NsFilesMetadataMigrationCommand extends AbstractCommand {
@Inject
private Provider<MetadataMigrationService> metadataMigrationServiceProvider;
@Override
public Integer call() throws Exception {
super.call();
try {
metadataMigrationServiceProvider.get().nsFilesMigration();
} catch (Exception e) {
System.err.println("❌ Namespace Files Metadata migration failed: " + e.getMessage());
e.printStackTrace();
return 1;
}
System.out.println("✅ Namespace Files Metadata migration complete.");
return 0;
}
}

View File

@@ -57,7 +57,7 @@ public class StateStoreMigrateCommand extends AbstractCommand {
String taskRunValue = statesUriPart.length > 2 ? statesUriPart[1] : null;
String stateSubName = statesUriPart[statesUriPart.length - 1];
boolean flowScoped = flowQualifierWithStateQualifiers[0].endsWith("/" + flow.getId());
StateStore stateStore = new StateStore(runContext(runContextFactory, flow), false);
StateStore stateStore = new StateStore(runContextFactory.of(flow, Map.of()), false);
try (InputStream is = storageInterface.get(flow.getTenantId(), flow.getNamespace(), stateStoreFileUri)) {
stateStore.putState(flowScoped, stateName, stateSubName, taskRunValue, is.readAllBytes());
@@ -70,12 +70,4 @@ public class StateStoreMigrateCommand extends AbstractCommand {
stdOut("Successfully ran the state-store migration.");
return 0;
}
private RunContext runContext(RunContextFactory runContextFactory, Flow flow) {
Map<String, String> flowVariables = new HashMap<>();
flowVariables.put("tenantId", flow.getTenantId());
flowVariables.put("id", flow.getId());
flowVariables.put("namespace", flow.getNamespace());
return runContextFactory.of(flow, Map.of("flow", flowVariables));
}
}

View File

@@ -0,0 +1,57 @@
package io.kestra.cli.commands.migrations.metadata;
import io.kestra.core.repositories.FlowRepositoryInterface;
import io.kestra.core.tenant.TenantService;
import io.kestra.core.utils.NamespaceUtils;
import io.kestra.core.utils.TestsUtils;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
public class MetadataMigrationServiceTest<T extends MetadataMigrationService> {
private static final String TENANT_ID = TestsUtils.randomTenant();
protected static final String SYSTEM_NAMESPACE = "my.system.namespace";
@Test
void namespacesPerTenant() {
Map<String, List<String>> expected = getNamespacesPerTenant();
Map<String, List<String>> result = metadataMigrationService(
expected
).namespacesPerTenant();
assertThat(result).hasSize(expected.size());
expected.forEach((tenantId, namespaces) -> {
assertThat(result.get(tenantId)).containsExactlyInAnyOrderElementsOf(
Stream.concat(
Stream.of(SYSTEM_NAMESPACE),
namespaces.stream()
).map(NamespaceUtils::asTree).flatMap(Collection::stream).distinct().toList()
);
});
}
protected Map<String, List<String>> getNamespacesPerTenant() {
return Map.of(TENANT_ID, List.of("my.first.namespace", "my.second.namespace", "another.namespace"));
}
protected T metadataMigrationService(Map<String, List<String>> namespacesPerTenant) {
FlowRepositoryInterface mockedFlowRepository = Mockito.mock(FlowRepositoryInterface.class);
Mockito.doAnswer((params) -> namespacesPerTenant.get(params.getArgument(0).toString())).when(mockedFlowRepository).findDistinctNamespace(Mockito.anyString());
NamespaceUtils namespaceUtils = Mockito.mock(NamespaceUtils.class);
Mockito.when(namespaceUtils.getSystemFlowNamespace()).thenReturn(SYSTEM_NAMESPACE);
//noinspection unchecked
return ((T) new MetadataMigrationService(mockedFlowRepository, new TenantService() {
@Override
public String resolveTenant() {
return TENANT_ID;
}
}, null, null, null, namespaceUtils));
}
}

View File

@@ -0,0 +1,175 @@
package io.kestra.cli.commands.migrations.metadata;
import io.kestra.cli.App;
import io.kestra.core.exceptions.ResourceExpiredException;
import io.kestra.core.models.flows.Flow;
import io.kestra.core.models.flows.GenericFlow;
import io.kestra.core.models.kv.PersistedKvMetadata;
import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
import io.kestra.core.repositories.FlowRepositoryInterface;
import io.kestra.core.repositories.KvMetadataRepositoryInterface;
import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
import io.kestra.core.serializers.JacksonMapper;
import io.kestra.core.storages.*;
import io.kestra.core.storages.kv.*;
import io.kestra.core.tenant.TenantService;
import io.kestra.core.utils.TestsUtils;
import io.kestra.plugin.core.log.Log;
import io.micronaut.configuration.picocli.PicocliRunner;
import io.micronaut.context.ApplicationContext;
import io.micronaut.context.env.Environment;
import io.micronaut.core.annotation.NonNull;
import org.junit.jupiter.api.Test;
import java.io.*;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class NsFilesMetadataMigrationCommandTest {
@Test
void run() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
System.setOut(new PrintStream(out));
ByteArrayOutputStream err = new ByteArrayOutputStream();
System.setErr(new PrintStream(err));
try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
/* Initial setup:
* - namespace 1: my/path, value
* - namespace 1: another/path
* - namespace 2: yet/another/path
* - Nothing in database */
String namespace = TestsUtils.randomNamespace();
String path = "/my/path";
StorageInterface storage = ctx.getBean(StorageInterface.class);
String value = "someValue";
putOldNsFile(storage, namespace, path, value);
String anotherPath = "/another/path";
String anotherValue = "anotherValue";
putOldNsFile(storage, namespace, anotherPath, anotherValue);
String anotherNamespace = TestsUtils.randomNamespace();
String yetAnotherPath = "/yet/another/path";
String yetAnotherValue = "yetAnotherValue";
putOldNsFile(storage, anotherNamespace, yetAnotherPath, yetAnotherValue);
NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository = ctx.getBean(NamespaceFileMetadataRepositoryInterface.class);
String tenantId = TenantService.MAIN_TENANT;
assertThat(namespaceFileMetadataRepository.findByPath(tenantId, namespace, path).isPresent()).isFalse();
/* Expected outcome from the migration command:
* - no namespace files has been migrated because no flow exist in the namespace so they are not picked up because we don't know they exist */
String[] nsFilesMetadataMigrationCommand = {
"migrate", "metadata", "nsfiles"
};
PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
// Still it's not in the metadata repository because no flow exist to find that namespace file
assertThat(namespaceFileMetadataRepository.findByPath(tenantId, namespace, path).isPresent()).isFalse();
assertThat(namespaceFileMetadataRepository.findByPath(tenantId, namespace, anotherPath).isPresent()).isFalse();
assertThat(namespaceFileMetadataRepository.findByPath(tenantId, anotherNamespace, yetAnotherPath).isPresent()).isFalse();
// A flow is created from namespace 1, so the namespace files in this namespace should be migrated
FlowRepositoryInterface flowRepository = ctx.getBean(FlowRepositoryInterface.class);
flowRepository.create(GenericFlow.of(Flow.builder()
.tenantId(tenantId)
.id("a-flow")
.namespace(namespace)
.tasks(List.of(Log.builder().id("log").type(Log.class.getName()).message("logging").build()))
.build()));
/* We run the migration again:
* - namespace 1 my/path file is seen and metadata is migrated to database
* - namespace 1 another/path file is seen and metadata is migrated to database
* - namespace 2 yet/another/path is not seen because no flow exist in this namespace */
out.reset();
PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
Optional<NamespaceFileMetadata> foundNsFile = namespaceFileMetadataRepository.findByPath(tenantId, namespace, path);
assertThat(foundNsFile.isPresent()).isTrue();
assertThat(foundNsFile.get().getVersion()).isEqualTo(1);
assertThat(foundNsFile.get().getSize()).isEqualTo(value.length());
Optional<NamespaceFileMetadata> anotherFoundNsFile = namespaceFileMetadataRepository.findByPath(tenantId, namespace, anotherPath);
assertThat(anotherFoundNsFile.isPresent()).isTrue();
assertThat(anotherFoundNsFile.get().getVersion()).isEqualTo(1);
assertThat(anotherFoundNsFile.get().getSize()).isEqualTo(anotherValue.length());
NamespaceFactory namespaceFactory = ctx.getBean(NamespaceFactory.class);
Namespace namespaceStorage = namespaceFactory.of(tenantId, namespace, storage);
FileAttributes nsFileRawMetadata = namespaceStorage.getFileMetadata(Path.of(path));
assertThat(nsFileRawMetadata.getSize()).isEqualTo(value.length());
assertThat(new String(namespaceStorage.getFileContent(Path.of(path)).readAllBytes())).isEqualTo(value);
FileAttributes anotherNsFileRawMetadata = namespaceStorage.getFileMetadata(Path.of(anotherPath));
assertThat(anotherNsFileRawMetadata.getSize()).isEqualTo(anotherValue.length());
assertThat(new String(namespaceStorage.getFileContent(Path.of(anotherPath)).readAllBytes())).isEqualTo(anotherValue);
assertThat(namespaceFileMetadataRepository.findByPath(tenantId, anotherNamespace, yetAnotherPath).isPresent()).isFalse();
assertThatThrownBy(() -> namespaceStorage.getFileMetadata(Path.of(yetAnotherPath))).isInstanceOf(FileNotFoundException.class);
/* We run one last time the migration without any change to verify that we don't resave an existing metadata.
* It covers the case where user didn't perform the migrate command yet but they played and added some KV from the UI (so those ones will already be in metadata database). */
out.reset();
PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
foundNsFile = namespaceFileMetadataRepository.findByPath(tenantId, namespace, path);
assertThat(foundNsFile.get().getVersion()).isEqualTo(1);
}
}
@Test
void namespaceWithoutNsFile() {
ByteArrayOutputStream out = new ByteArrayOutputStream();
System.setOut(new PrintStream(out));
ByteArrayOutputStream err = new ByteArrayOutputStream();
System.setErr(new PrintStream(err));
try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
String tenantId = TenantService.MAIN_TENANT;
String namespace = TestsUtils.randomNamespace();
// A flow is created from namespace 1, so the namespace files in this namespace should be migrated
FlowRepositoryInterface flowRepository = ctx.getBean(FlowRepositoryInterface.class);
flowRepository.create(GenericFlow.of(Flow.builder()
.tenantId(tenantId)
.id("a-flow")
.namespace(namespace)
.tasks(List.of(Log.builder().id("log").type(Log.class.getName()).message("logging").build()))
.build()));
String[] nsFilesMetadataMigrationCommand = {
"migrate", "metadata", "nsfiles"
};
PicocliRunner.call(App.class, ctx, nsFilesMetadataMigrationCommand);
assertThat(out.toString()).contains("✅ Namespace Files Metadata migration complete.");
assertThat(err.toString()).doesNotContain("java.nio.file.NoSuchFileException");
}
}
private static void putOldNsFile(StorageInterface storage, String namespace, String path, String value) throws IOException {
URI nsFileStorageUri = getNsFileStorageUri(namespace, path);
storage.put(TenantService.MAIN_TENANT, namespace, nsFileStorageUri, new StorageObject(
null,
new ByteArrayInputStream(value.getBytes(StandardCharsets.UTF_8))
));
}
private static @NonNull URI getNsFileStorageUri(String namespace, String path) {
return URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.namespaceFilePrefix(namespace) + path);
}
}

View File

@@ -55,11 +55,7 @@ class StateStoreMigrateCommandTest {
);
assertThat(storage.exists(tenantId, flow.getNamespace(), oldStateStoreUri)).isTrue();
RunContext runContext = ctx.getBean(RunContextFactory.class).of(flow, Map.of("flow", Map.of(
"tenantId", tenantId,
"id", flow.getId(),
"namespace", flow.getNamespace()
)));
RunContext runContext = ctx.getBean(RunContextFactory.class).of(flow, Map.of());
StateStore stateStore = new StateStore(runContext, true);
Assertions.assertThrows(MigrationRequiredException.class, () -> stateStore.getState(true, "my-state", "sub-name", "my-taskrun-value"));

View File

@@ -19,7 +19,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junitpioneer.jupiter.RetryingTest;
import static io.kestra.core.utils.Rethrow.throwRunnable;
import static org.assertj.core.api.Assertions.assertThat;
@@ -59,7 +58,7 @@ class FileChangedEventListenerTest {
}
@FlakyTest
@RetryingTest(2)
@Test
void test() throws IOException, TimeoutException {
var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getSimpleName(), "test");
// remove the flow if it already exists
@@ -98,7 +97,7 @@ class FileChangedEventListenerTest {
}
@FlakyTest
@RetryingTest(2)
@Test
void testWithPluginDefault() throws IOException, TimeoutException {
var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getName(), "testWithPluginDefault");
// remove the flow if it already exists
@@ -138,4 +137,4 @@ class FileChangedEventListenerTest {
Duration.ofSeconds(10)
);
}
}
}

View File

@@ -21,6 +21,7 @@ kestra:
server:
liveness:
enabled: false
termination-grace-period: 5s
micronaut:
http:
services:

View File

@@ -15,6 +15,7 @@ import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -84,6 +85,11 @@ public abstract class KestraContext {
public abstract StorageInterface getStorageInterface();
/**
* Returns the Micronaut active environments.
*/
public abstract Set<String> getEnvironments();
/**
* Shutdowns the Kestra application.
*/
@@ -182,5 +188,10 @@ public abstract class KestraContext {
// Lazy init of the PluginRegistry.
return this.applicationContext.getBean(StorageInterface.class);
}
@Override
public Set<String> getEnvironments() {
return this.applicationContext.getEnvironment().getActiveNames();
}
}
}

View File

@@ -0,0 +1,23 @@
package io.kestra.core.exceptions;
/**
* Exception that can be thrown when a Flow is not found.
*/
public class FlowNotFoundException extends NotFoundException {
/**
* Creates a new {@link FlowNotFoundException} instance.
*/
public FlowNotFoundException() {
super();
}
/**
* Creates a new {@link NotFoundException} instance.
*
* @param message the error message.
*/
public FlowNotFoundException(final String message) {
super(message);
}
}

View File

@@ -0,0 +1,15 @@
package io.kestra.core.exceptions;
import java.io.Serial;
public class ResourceAccessDeniedException extends KestraRuntimeException {
@Serial
private static final long serialVersionUID = 1L;
public ResourceAccessDeniedException() {
}
public ResourceAccessDeniedException(String message) {
super(message);
}
}

View File

@@ -7,7 +7,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
@@ -65,7 +64,7 @@ public interface HasSource {
if (isYAML(fileName)) {
byte[] bytes = inputStream.readAllBytes();
List<String> sources = List.of(new String(bytes).split("---"));
List<String> sources = List.of(new String(bytes).split("(?m)^---\\s*$"));
for (int i = 0; i < sources.size(); i++) {
String source = sources.get(i);
reader.accept(source, String.valueOf(i));

View File

@@ -4,13 +4,16 @@ import io.kestra.core.utils.MapUtils;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.annotation.Nullable;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.Pattern;
import java.util.*;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@Schema(description = "A key/value pair that can be attached to a Flow or Execution. Labels are often used to organize and categorize objects.")
public record Label(@NotEmpty String key, @NotEmpty String value) {
public record Label(
@NotEmpty @Pattern(regexp = "^[\\p{Ll}][\\p{L}0-9._-]*$", message = "Invalid label key. A valid key contains only lowercase letters numbers hyphens (-) underscores (_) or periods (.) and must begin with a lowercase letter.") String key,
@NotEmpty String value) {
public static final String SYSTEM_PREFIX = "system.";
// system labels

View File

@@ -94,7 +94,7 @@ public record QueryFilter(
KIND("kind") {
@Override
public List<Op> supportedOp() {
return List.of(Op.EQUALS,Op.NOT_EQUALS);
return List.of(Op.EQUALS,Op.NOT_EQUALS, Op.IN, Op.NOT_IN);
}
},
LABELS("labels") {
@@ -106,7 +106,7 @@ public record QueryFilter(
FLOW_ID("flowId") {
@Override
public List<Op> supportedOp() {
return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX);
return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX, Op.IN, Op.NOT_IN, Op.PREFIX);
}
},
UPDATED("updated") {
@@ -180,6 +180,24 @@ public record QueryFilter(
public List<Op> supportedOp() {
return List.of(Op.EQUALS, Op.NOT_EQUALS);
}
},
PATH("path") {
@Override
public List<Op> supportedOp() {
return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.IN);
}
},
PARENT_PATH("parentPath") {
@Override
public List<Op> supportedOp() {
return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.STARTS_WITH);
}
},
VERSION("version") {
@Override
public List<Op> supportedOp() {
return List.of(Op.EQUALS, Op.NOT_EQUALS);
}
};
private static final Map<String, Field> BY_VALUE = Arrays.stream(values())
@@ -208,7 +226,7 @@ public record QueryFilter(
FLOW {
@Override
public List<Field> supportedField() {
return List.of(Field.LABELS, Field.NAMESPACE, Field.QUERY, Field.SCOPE);
return List.of(Field.LABELS, Field.NAMESPACE, Field.QUERY, Field.SCOPE, Field.FLOW_ID);
}
},
NAMESPACE {
@@ -223,7 +241,7 @@ public record QueryFilter(
return List.of(
Field.QUERY, Field.SCOPE, Field.FLOW_ID, Field.START_DATE, Field.END_DATE,
Field.STATE, Field.LABELS, Field.TRIGGER_EXECUTION_ID, Field.CHILD_FILTER,
Field.NAMESPACE,Field.KIND
Field.NAMESPACE, Field.KIND
);
}
},
@@ -275,6 +293,19 @@ public record QueryFilter(
Field.UPDATED
);
}
},
NAMESPACE_FILE_METADATA {
@Override
public List<Field> supportedField() {
return List.of(
Field.QUERY,
Field.NAMESPACE,
Field.PATH,
Field.PARENT_PATH,
Field.VERSION,
Field.UPDATED
);
}
};
public abstract List<Field> supportedField();

View File

@@ -658,18 +658,20 @@ public class Execution implements DeletedInterface, TenantInterface {
public boolean hasFailedNoRetry(List<ResolvedTask> resolvedTasks, TaskRun parentTaskRun) {
return this.findTaskRunByTasks(resolvedTasks, parentTaskRun)
.stream()
.anyMatch(taskRun -> {
ResolvedTask resolvedTask = resolvedTasks.stream()
.filter(t -> t.getTask().getId().equals(taskRun.getTaskId())).findFirst()
.orElse(null);
if (resolvedTask == null) {
log.warn("Can't find task for taskRun '{}' in parentTaskRun '{}'",
taskRun.getId(), parentTaskRun.getId());
return false;
}
return !taskRun.shouldBeRetried(resolvedTask.getTask().getRetry())
&& taskRun.getState().isFailed();
});
// NOTE: we check on isFailed first to avoid the costly shouldBeRetried() method
.anyMatch(taskRun -> taskRun.getState().isFailed() && shouldNotBeRetried(resolvedTasks, parentTaskRun, taskRun));
}
private static boolean shouldNotBeRetried(List<ResolvedTask> resolvedTasks, TaskRun parentTaskRun, TaskRun taskRun) {
ResolvedTask resolvedTask = resolvedTasks.stream()
.filter(t -> t.getTask().getId().equals(taskRun.getTaskId())).findFirst()
.orElse(null);
if (resolvedTask == null) {
log.warn("Can't find task for taskRun '{}' in parentTaskRun '{}'",
taskRun.getId(), parentTaskRun.getId());
return false;
}
return !taskRun.shouldBeRetried(resolvedTask.getTask().getRetry());
}
public boolean hasCreated() {

View File

@@ -1,15 +1,16 @@
package io.kestra.core.models.executions;
import io.micronaut.core.annotation.Introspected;
import lombok.Builder;
import lombok.Value;
import io.kestra.core.models.tasks.Output;
import io.kestra.core.models.triggers.AbstractTrigger;
import io.micronaut.core.annotation.Introspected;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotNull;
import lombok.Builder;
import lombok.Value;
import java.net.URI;
import java.util.Collections;
import java.util.Map;
import jakarta.validation.constraints.NotNull;
@Value
@Builder
@@ -21,6 +22,7 @@ public class ExecutionTrigger {
@NotNull
String type;
@Schema(type = "object", additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
Map<String, Object> variables;
URI logFile;

View File

@@ -314,4 +314,11 @@ public class TaskRun implements TenantInterface {
.build();
}
public TaskRun addAttempt(TaskRunAttempt attempt) {
if (this.attempts == null) {
this.attempts = new ArrayList<>();
}
this.attempts.add(attempt);
return this;
}
}

View File

@@ -24,4 +24,8 @@ public class Concurrency {
public enum Behavior {
QUEUE, CANCEL, FAIL;
}
public static boolean possibleTransitions(State.Type type) {
return type.equals(State.Type.CANCELLED) || type.equals(State.Type.FAILED);
}
}

View File

@@ -11,6 +11,7 @@ import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
import io.kestra.core.exceptions.InternalException;
import io.kestra.core.models.HasUID;
import io.kestra.core.models.annotations.PluginProperty;
import io.kestra.core.models.flows.check.Check;
import io.kestra.core.models.flows.sla.SLA;
import io.kestra.core.models.listeners.Listener;
import io.kestra.core.models.tasks.FlowableTask;
@@ -129,6 +130,14 @@ public class Flow extends AbstractFlow implements HasUID {
@Valid
@PluginProperty
List<SLA> sla;
@Schema(
title = "Conditions evaluated before the flow is executed.",
description = "A list of conditions that are evaluated before the flow is executed. If no checks are defined, the flow executes normally."
)
@Valid
@PluginProperty
List<Check> checks;
public Stream<String> allTypes() {
return Stream.of(

View File

@@ -43,6 +43,7 @@ public class FlowWithSource extends Flow {
.concurrency(this.concurrency)
.retry(this.retry)
.sla(this.sla)
.checks(this.checks)
.build();
}
@@ -85,6 +86,7 @@ public class FlowWithSource extends Flow {
.concurrency(flow.concurrency)
.retry(flow.retry)
.sla(flow.sla)
.checks(flow.checks)
.build();
}
}

View File

@@ -84,12 +84,24 @@ public class State {
);
}
/**
* non-terminated execution duration is hard to provide in SQL, so we set it to null when endDate is empty
*/
@JsonProperty(access = JsonProperty.Access.READ_ONLY)
public Duration getDuration() {
return Duration.between(
this.histories.getFirst().getDate(),
this.histories.size() > 1 ? this.histories.get(this.histories.size() - 1).getDate() : Instant.now()
);
@JsonInclude(JsonInclude.Include.NON_EMPTY)
public Optional<Duration> getDuration() {
if (this.getEndDate().isPresent()) {
return Optional.of(Duration.between(this.getStartDate(), this.getEndDate().get()));
} else {
return Optional.empty();
}
}
/**
* @return either the Duration persisted in database, or calculate it on the fly for non-terminated executions
*/
public Duration getDurationOrComputeIt() {
return this.getDuration().orElseGet(() -> Duration.between(this.getStartDate(), Instant.now()));
}
@JsonProperty(access = JsonProperty.Access.READ_ONLY)
@@ -109,7 +121,7 @@ public class State {
public String humanDuration() {
try {
return DurationFormatUtils.formatDurationHMS(getDuration().toMillis());
return DurationFormatUtils.formatDurationHMS(getDurationOrComputeIt().toMillis());
} catch (Throwable e) {
return getDuration().toString();
}
@@ -255,6 +267,10 @@ public class State {
return this == Type.RUNNING || this == Type.KILLING;
}
public boolean onlyRunning() {
return this == Type.RUNNING;
}
public boolean isFailed() {
return this == Type.FAILED;
}

View File

@@ -0,0 +1,109 @@
package io.kestra.core.models.flows.check;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotNull;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
/**
* Represents a check within a Kestra flow.
* <p>
* A {@code Check} defines a boolean condition that is evaluated when validating flow's inputs
* and before triggering an execution.
* <p>
* If the condition evaluates to {@code false}, the configured {@link Behavior}
* determines how the execution proceeds, and the {@link Style} determines how
* the message is visually presented in the UI.
* </p>
*/
@SuperBuilder
@Getter
@NoArgsConstructor
public class Check {
/**
* The condition to evaluate.
*/
@NotNull
@NotEmpty
String condition;
/**
* The message associated with this check, will be displayed when the condition evaluates to {@code false}.
*/
@NotEmpty
String message;
/**
* Defines the style of the message displayed in the UI when the condition evaluates to {@code false}.
*/
Style style = Style.INFO;
/**
* The behavior to apply when the condition evaluates to {@code false}.
*/
Behavior behavior = Behavior.BLOCK_EXECUTION;
/**
* The visual style used to display the message when the check fails.
*/
public enum Style {
/**
* Display the message as an error.
*/
ERROR,
/**
* Display the message as a success indicator.
*/
SUCCESS,
/**
* Display the message as a warning.
*/
WARNING,
/**
* Display the message as informational content.
*/
INFO;
}
/**
* Defines how the flow should behave when the condition evaluates to {@code false}.
*/
public enum Behavior {
/**
* Block the creation of the execution.
*/
BLOCK_EXECUTION,
/**
* Create the execution as failed.
*/
FAIL_EXECUTION,
/**
* Create a new execution as a result of the check failing.
*/
CREATE_EXECUTION;
}
/**
* Resolves the effective behavior for a list of {@link Check}s based on priority.
*
* @param checks the list of checks whose behaviors are to be evaluated
* @return the highest-priority behavior, or {@code CREATE_EXECUTION} if the list is empty or only contains nulls
*/
public static Check.Behavior resolveBehavior(List<Check> checks) {
if (checks == null || checks.isEmpty()) {
return Behavior.CREATE_EXECUTION;
}
return checks.stream()
.map(Check::getBehavior)
.filter(Objects::nonNull).min(Comparator.comparingInt(Enum::ordinal))
.orElse(Behavior.CREATE_EXECUTION);
}
}

View File

@@ -20,7 +20,6 @@ import java.util.Optional;
@Slf4j
@Getter
@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
@AllArgsConstructor
@ToString
@EqualsAndHashCode
public class PersistedKvMetadata implements DeletedInterface, TenantInterface, HasUID {
@@ -54,6 +53,19 @@ public class PersistedKvMetadata implements DeletedInterface, TenantInterface, H
private boolean deleted;
public PersistedKvMetadata(String tenantId, String namespace, String name, String description, Integer version, boolean last, @Nullable Instant expirationDate, @Nullable Instant created, @Nullable Instant updated, boolean deleted) {
this.tenantId = tenantId;
this.namespace = namespace;
this.name = name;
this.description = description;
this.version = version;
this.last = last;
this.expirationDate = expirationDate;
this.created = Optional.ofNullable(created).orElse(Instant.now());
this.updated = updated;
this.deleted = deleted;
}
public static PersistedKvMetadata from(String tenantId, KVEntry kvEntry) {
return PersistedKvMetadata.builder()
.tenantId(tenantId)
@@ -68,12 +80,15 @@ public class PersistedKvMetadata implements DeletedInterface, TenantInterface, H
}
public PersistedKvMetadata asLast() {
Instant saveDate = Instant.now();
return this.toBuilder().created(Optional.ofNullable(this.created).orElse(saveDate)).updated(saveDate).last(true).build();
return this.toBuilder().updated(Instant.now()).last(true).build();
}
public PersistedKvMetadata toDeleted() {
return this.toBuilder().updated(Instant.now()).deleted(true).build();
}
@Override
public String uid() {
return IdUtils.fromParts(getTenantId(), getNamespace(), getName(), getVersion().toString());
return IdUtils.fromParts(getTenantId(), getNamespace(), getName(), String.valueOf(getVersion()));
}
}

View File

@@ -0,0 +1,132 @@
package io.kestra.core.models.namespaces.files;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import io.kestra.core.models.DeletedInterface;
import io.kestra.core.models.HasUID;
import io.kestra.core.models.TenantInterface;
import io.kestra.core.storages.FileAttributes;
import io.kestra.core.storages.NamespaceFile;
import io.kestra.core.utils.IdUtils;
import io.swagger.v3.oas.annotations.Hidden;
import jakarta.annotation.Nullable;
import jakarta.validation.constraints.NotNull;
import jakarta.validation.constraints.Pattern;
import lombok.*;
import lombok.experimental.FieldDefaults;
import lombok.extern.slf4j.Slf4j;
import java.time.Instant;
@Builder(toBuilder = true)
@Slf4j
@Getter
@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
@ToString
@EqualsAndHashCode
public class NamespaceFileMetadata implements DeletedInterface, TenantInterface, HasUID {
@With
@Hidden
@Pattern(regexp = "^[a-z0-9][a-z0-9_-]*")
private String tenantId;
@NotNull
private String namespace;
@NotNull
private String path;
private String parentPath;
@NotNull
private Integer version;
@Builder.Default
private boolean last = true;
@NotNull
private Long size;
@Builder.Default
private Instant created = Instant.now();
@Nullable
private Instant updated;
private boolean deleted;
@JsonCreator
public NamespaceFileMetadata(String tenantId, String namespace, String path, String parentPath, Integer version, boolean last, Long size, Instant created, @Nullable Instant updated, boolean deleted) {
this.tenantId = tenantId;
this.namespace = namespace;
this.path = path;
this.parentPath = parentPath(path);
this.version = version;
this.last = last;
this.size = size;
this.created = created;
this.updated = updated;
this.deleted = deleted;
}
public static String path(String path, boolean trailingSlash) {
if (trailingSlash && !path.endsWith("/")) {
return path + "/";
} else if (!trailingSlash && path.endsWith("/")) {
return path.substring(0, path.length() - 1);
}
return path;
}
public String path(boolean trailingSlash) {
return path(this.path, trailingSlash);
}
public static String parentPath(String path) {
String withoutTrailingSlash = path.endsWith("/") ? path.substring(0, path.length() - 1) : path;
// The parent path can't be set, it's always computed
return withoutTrailingSlash.contains("/") ?
withoutTrailingSlash.substring(0, withoutTrailingSlash.lastIndexOf("/") + 1) :
null;
}
public static NamespaceFileMetadata of(String tenantId, NamespaceFile namespaceFile) {
return NamespaceFileMetadata.builder()
.tenantId(tenantId)
.namespace(namespaceFile.namespace())
.path(namespaceFile.path(true).toString())
.version(namespaceFile.version())
.build();
}
public static NamespaceFileMetadata of(String tenantId, String namespace, String path, FileAttributes fileAttributes) {
return NamespaceFileMetadata.builder()
.tenantId(tenantId)
.namespace(namespace)
.path(path)
.created(Instant.ofEpochMilli(fileAttributes.getCreationTime()))
.updated(Instant.ofEpochMilli(fileAttributes.getLastModifiedTime()))
.size(fileAttributes.getSize())
.version(1)
.build();
}
public NamespaceFileMetadata asLast() {
Instant saveDate = Instant.now();
return this.toBuilder().updated(saveDate).last(true).build();
}
public NamespaceFileMetadata toDeleted() {
return this.toBuilder().deleted(true).updated(Instant.now()).build();
}
@Override
public String uid() {
return IdUtils.fromParts(getTenantId(), getNamespace(), getPath(), String.valueOf(getVersion()));
}
@JsonIgnore
public boolean isDirectory() {
return this.path.endsWith("/");
}
}

View File

@@ -35,7 +35,6 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
@JsonDeserialize(using = Property.PropertyDeserializer.class)
@JsonSerialize(using = Property.PropertySerializer.class)
@Builder
@NoArgsConstructor
@AllArgsConstructor(access = AccessLevel.PACKAGE)
@Schema(
oneOf = {
@@ -51,6 +50,7 @@ public class Property<T> {
.copy()
.configure(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS, false);
private final boolean skipCache;
private String expression;
private T value;
@@ -60,13 +60,23 @@ public class Property<T> {
@Deprecated
// Note: when not used, this constructor would not be deleted but made private so it can only be used by ofExpression(String) and the deserializer
public Property(String expression) {
this.expression = expression;
this(expression, false);
}
private Property(String expression, boolean skipCache) {
this.expression = expression;
this.skipCache = skipCache;
}
/**
* @deprecated use {@link #ofValue(Object)} instead.
*/
@VisibleForTesting
@Deprecated
public Property(Map<?, ?> map) {
try {
expression = MAPPER.writeValueAsString(map);
this.skipCache = false;
} catch (JsonProcessingException e) {
throw new IllegalArgumentException(e);
}
@@ -79,14 +89,11 @@ public class Property<T> {
/**
* Returns a new {@link Property} with no cached rendered value,
* so that the next render will evaluate its original Pebble expression.
* <p>
* The returned property will still cache its rendered result.
* To re-evaluate on a subsequent render, call {@code skipCache()} again.
*
* @return a new {@link Property} without a pre-rendered value
*/
public Property<T> skipCache() {
return Property.ofExpression(expression);
return new Property<>(expression, true);
}
/**
@@ -133,6 +140,7 @@ public class Property<T> {
/**
* Build a new Property object with a Pebble expression.<br>
* This property object will not cache its rendered value.
* <p>
* Use {@link #ofValue(Object)} to build a property with a value instead.
*/
@@ -142,11 +150,11 @@ public class Property<T> {
throw new IllegalArgumentException("'expression' must be a valid Pebble expression");
}
return new Property<>(expression);
return new Property<>(expression, true);
}
/**
* Render a property then convert it to its target type.<br>
* Render a property, then convert it to its target type.<br>
* <p>
* This method is designed to be used only by the {@link io.kestra.core.runners.RunContextProperty}.
*
@@ -164,7 +172,7 @@ public class Property<T> {
* @see io.kestra.core.runners.RunContextProperty#as(Class, Map)
*/
public static <T> T as(Property<T> property, PropertyContext context, Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
if (property.value == null) {
if (property.skipCache || property.value == null) {
String rendered = context.render(property.expression, variables);
property.value = MAPPER.convertValue(rendered, clazz);
}
@@ -192,7 +200,7 @@ public class Property<T> {
*/
@SuppressWarnings("unchecked")
public static <T, I> T asList(Property<T> property, PropertyContext context, Class<I> itemClazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
if (property.value == null) {
if (property.skipCache || property.value == null) {
JavaType type = MAPPER.getTypeFactory().constructCollectionLikeType(List.class, itemClazz);
try {
String trimmedExpression = property.expression.trim();
@@ -244,7 +252,7 @@ public class Property<T> {
*/
@SuppressWarnings({"rawtypes", "unchecked"})
public static <T, K, V> T asMap(Property<T> property, RunContext runContext, Class<K> keyClass, Class<V> valueClass, Map<String, Object> variables) throws IllegalVariableEvaluationException {
if (property.value == null) {
if (property.skipCache || property.value == null) {
JavaType targetMapType = MAPPER.getTypeFactory().constructMapType(Map.class, keyClass, valueClass);
try {

View File

@@ -4,10 +4,8 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
import io.kestra.core.models.tasks.runners.TaskLogLineMatcher.TaskLogMatch;
import io.kestra.core.runners.DefaultRunContext;
import io.kestra.core.runners.RunContext;
import io.kestra.core.serializers.JacksonMapper;
import io.kestra.core.services.FlowService;
import jakarta.validation.constraints.NotNull;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
@@ -38,6 +36,7 @@ import static io.kestra.core.utils.Rethrow.throwConsumer;
abstract public class PluginUtilsService {
private static final TypeReference<Map<String, String>> MAP_TYPE_REFERENCE = new TypeReference<>() {};
private static final TaskLogLineMatcher LOG_LINE_MATCHER = new TaskLogLineMatcher();
public static Map<String, String> createOutputFiles(
Path tempDirectory,
@@ -170,12 +169,9 @@ abstract public class PluginUtilsService {
}
public static Map<String, Object> parseOut(String line, Logger logger, RunContext runContext, boolean isStdErr, Instant customInstant) {
TaskLogLineMatcher logLineMatcher = ((DefaultRunContext) runContext).getApplicationContext().getBean(TaskLogLineMatcher.class);
Map<String, Object> outputs = new HashMap<>();
try {
Optional<TaskLogMatch> matches = logLineMatcher.matches(line, logger, runContext, customInstant);
Optional<TaskLogMatch> matches = LOG_LINE_MATCHER.matches(line, logger, runContext, customInstant);
if (matches.isPresent()) {
TaskLogMatch taskLogMatch = matches.get();
outputs.putAll(taskLogMatch.outputs());
@@ -215,8 +211,7 @@ abstract public class PluginUtilsService {
realNamespace = runContext.render(namespace);
realFlowId = runContext.render(flowId);
// validate that the flow exists: a.k.a access is authorized by this namespace
FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);
flowService.checkAllowedNamespace(flowInfo.tenantId(), realNamespace, flowInfo.tenantId(), flowInfo.namespace());
runContext.acl().allowNamespace(realNamespace).check();
} else if (namespace != null || flowId != null) {
throw new IllegalArgumentException("Both `namespace` and `flowId` must be set when `executionId` is set.");
} else {

View File

@@ -27,7 +27,6 @@ import static io.kestra.core.runners.RunContextLogger.ORIGINAL_TIMESTAMP_KEY;
* ::{"outputs":{"key":"value"}}::
* }</pre>
*/
@Singleton
public class TaskLogLineMatcher {
protected static final Pattern LOG_DATA_SYNTAX = Pattern.compile("^::(\\{.*})::$");
@@ -108,4 +107,4 @@ public class TaskLogLineMatcher {
String message
) {
}
}
}

View File

@@ -2,7 +2,6 @@ package io.kestra.core.repositories;
import io.kestra.core.models.QueryFilter;
import io.kestra.core.models.executions.Execution;
import io.kestra.core.models.executions.TaskRun;
import io.kestra.core.models.executions.statistics.DailyExecutionStatistics;
import io.kestra.core.models.executions.statistics.ExecutionCount;
import io.kestra.core.models.executions.statistics.Flow;
@@ -94,6 +93,8 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
Flux<Execution> findAllAsync(@Nullable String tenantId);
Flux<Execution> findAsync(String tenantId, List<QueryFilter> filters);
Execution delete(Execution execution);
Integer purge(Execution execution);

View File

@@ -8,6 +8,7 @@ import io.kestra.plugin.core.dashboard.data.Flows;
import io.micronaut.data.model.Pageable;
import jakarta.annotation.Nullable;
import jakarta.validation.ConstraintViolationException;
import reactor.core.publisher.Flux;
import java.util.List;
import java.util.Optional;
@@ -158,6 +159,8 @@ public interface FlowRepositoryInterface extends QueryBuilderInterface<Flows.Fie
.toList();
}
Flux<Flow> findAsync(String tenantId, List<QueryFilter> filters);
FlowWithSource create(GenericFlow flow);
FlowWithSource update(GenericFlow flow, FlowInterface previous) throws ConstraintViolationException;

View File

@@ -0,0 +1,46 @@
package io.kestra.core.repositories;
import io.kestra.core.models.FetchVersion;
import io.kestra.core.models.QueryFilter;
import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
import io.micronaut.data.model.Pageable;
import java.io.IOException;
import java.util.List;
import java.util.Optional;
public interface NamespaceFileMetadataRepositoryInterface extends SaveRepositoryInterface<NamespaceFileMetadata> {
Optional<NamespaceFileMetadata> findByPath(
String tenantId,
String namespace,
String path
) throws IOException;
default ArrayListTotal<NamespaceFileMetadata> find(
Pageable pageable,
String tenantId,
List<QueryFilter> filters,
boolean allowDeleted
) {
return this.find(pageable, tenantId, filters, allowDeleted, FetchVersion.LATEST);
}
ArrayListTotal<NamespaceFileMetadata> find(
Pageable pageable,
String tenantId,
List<QueryFilter> filters,
boolean allowDeleted,
FetchVersion fetchBehavior
);
default NamespaceFileMetadata delete(NamespaceFileMetadata namespaceFileMetadata) throws IOException {
return this.save(namespaceFileMetadata.toBuilder().deleted(true).build());
}
/**
* Purge (hard delete) a list of namespace files metadata. If no version is specified, all versions are purged.
* @param namespaceFilesMetadata the list of namespace files metadata to purge
* @return the number of purged namespace files metadata
*/
Integer purge(List<NamespaceFileMetadata> namespaceFilesMetadata);
}

View File

@@ -43,9 +43,9 @@ public interface TriggerRepositoryInterface extends QueryBuilderInterface<Trigge
/**
* Find all triggers that match the query, return a flux of triggers
* as the search is not paginated
*/
Flux<Trigger> find(String tenantId, List<QueryFilter> filters);
Flux<Trigger> findAsync(String tenantId, List<QueryFilter> filters);
default Function<String, String> sortMapping() throws IllegalArgumentException {
return Function.identity();

View File

@@ -0,0 +1,50 @@
package io.kestra.core.runners;
import javax.annotation.CheckReturnValue;
import java.util.List;
/**
* Check if the current taskrun has access to the requested resources.
*
* <p>
* IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
*
* @see AllowedResources
*/
public interface AclChecker {
/**Tasks that need to access resources outside their namespace should use this interface to check ACL (Allowed namespaces in EE).
* Allow all namespaces.
* <p>
* IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
*/
@CheckReturnValue
AllowedResources allowAllNamespaces();
/**
* Allow only the given namespace.
* <p>
* IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
*/
@CheckReturnValue
AllowedResources allowNamespace(String namespace);
/**
* Allow only the given namespaces.
* <p>
* IMPORTANT: remember to call the <code>check()</code> method to check the ACL.
*/
@CheckReturnValue
AllowedResources allowNamespaces(List<String> namespaces);
/**
* Represents a set of allowed resources.
* Tasks that need to access resources outside their namespace should call the <code>check()</code> method to check the ACL (Allowed namespaces in EE).
*/
interface AllowedResources {
/**
* Check if the current taskrun has access to the requested resources.
*/
void check();
}
}

View File

@@ -0,0 +1,86 @@
package io.kestra.core.runners;
import io.kestra.core.services.NamespaceService;
import io.micronaut.context.ApplicationContext;
import java.util.List;
import java.util.Objects;
class AclCheckerImpl implements AclChecker {
private final NamespaceService namespaceService;
private final RunContext.FlowInfo flowInfo;
AclCheckerImpl(ApplicationContext applicationContext, RunContext.FlowInfo flowInfo) {
this.namespaceService = applicationContext.getBean(NamespaceService.class);
this.flowInfo = flowInfo;
}
@Override
public AllowedResources allowAllNamespaces() {
return new AllowAllNamespaces(flowInfo, namespaceService);
}
@Override
public AllowedResources allowNamespace(String namespace) {
return new AllowNamespace(flowInfo, namespaceService, namespace);
}
@Override
public AllowedResources allowNamespaces(List<String> namespaces) {
return new AllowNamespaces(flowInfo, namespaceService, namespaces);
}
static class AllowAllNamespaces implements AllowedResources {
private final RunContext.FlowInfo flowInfo;
private final NamespaceService namespaceService;
AllowAllNamespaces(RunContext.FlowInfo flowInfo, NamespaceService namespaceService) {
this.flowInfo = Objects.requireNonNull(flowInfo);
this.namespaceService = Objects.requireNonNull(namespaceService);
}
@Override
public void check() {
this.namespaceService.checkAllowedAllNamespaces(flowInfo.tenantId(), flowInfo.tenantId(), flowInfo.namespace());
}
}
static class AllowNamespace implements AllowedResources {
private final RunContext.FlowInfo flowInfo;
private final NamespaceService namespaceService;
private final String namespace;
public AllowNamespace(RunContext.FlowInfo flowInfo, NamespaceService namespaceService, String namespace) {
this.flowInfo = Objects.requireNonNull(flowInfo);
this.namespaceService = Objects.requireNonNull(namespaceService);
this.namespace = Objects.requireNonNull(namespace);
}
@Override
public void check() {
namespaceService.checkAllowedNamespace(flowInfo.tenantId(), namespace, flowInfo.tenantId(), flowInfo.namespace());
}
}
static class AllowNamespaces implements AllowedResources {
private final RunContext.FlowInfo flowInfo;
private final NamespaceService namespaceService;
private final List<String> namespaces;
AllowNamespaces(RunContext.FlowInfo flowInfo, NamespaceService namespaceService, List<String> namespaces) {
this.flowInfo = Objects.requireNonNull(flowInfo);
this.namespaceService = Objects.requireNonNull(namespaceService);
this.namespaces = Objects.requireNonNull(namespaces);
if (namespaces.isEmpty()) {
throw new IllegalArgumentException("At least one namespace must be provided");
}
}
@Override
public void check() {
namespaces.forEach(namespace -> namespaceService.checkAllowedNamespace(flowInfo.tenantId(), namespace, flowInfo.tenantId(), flowInfo.namespace()));
}
}
}

View File

@@ -123,7 +123,12 @@ public class DefaultRunContext extends RunContext {
this.traceParent = traceParent;
}
/**
* @deprecated Plugin should not use the ApplicationContext anymore, and neither should they cast to this implementation.
* Plugin should instead rely on supported API only.
*/
@JsonIgnore
@Deprecated(since = "1.2.0", forRemoval = true)
public ApplicationContext getApplicationContext() {
return applicationContext;
}
@@ -574,6 +579,11 @@ public class DefaultRunContext extends RunContext {
return isInitialized.get();
}
@Override
public AclChecker acl() {
return new AclCheckerImpl(this.applicationContext, flowInfo());
}
@Override
public LocalPath localPath() {
return localPath;

View File

@@ -53,12 +53,10 @@ public final class ExecutableUtils {
}
public static SubflowExecutionResult subflowExecutionResult(TaskRun parentTaskrun, Execution execution) {
List<TaskRunAttempt> attempts = parentTaskrun.getAttempts() == null ? new ArrayList<>() : new ArrayList<>(parentTaskrun.getAttempts());
attempts.add(TaskRunAttempt.builder().state(parentTaskrun.getState()).build());
return SubflowExecutionResult.builder()
.executionId(execution.getId())
.state(parentTaskrun.getState().getCurrent())
.parentTaskRun(parentTaskrun.withAttempts(attempts))
.parentTaskRun(parentTaskrun.addAttempt(TaskRunAttempt.builder().state(parentTaskrun.getState()).build()))
.build();
}

View File

@@ -158,11 +158,7 @@ public class FlowInputOutput {
File tempFile = File.createTempFile(prefix, fileExtension);
try (var inputStream = fileUpload.getInputStream();
var outputStream = new FileOutputStream(tempFile)) {
long transferredBytes = inputStream.transferTo(outputStream);
if (transferredBytes == 0) {
sink.error(new KestraRuntimeException("Can't upload file: " + fileUpload.getFilename()));
return;
}
inputStream.transferTo(outputStream);
URI from = storageInterface.from(execution, inputId, fileName, tempFile);
sink.next(Map.entry(inputId, from.toString()));
} finally {
@@ -382,11 +378,11 @@ public class FlowInputOutput {
@SuppressWarnings("unchecked")
private static <T> Object resolveDefaultPropertyAs(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
return Property.as((Property<T>) input.getDefaults(), renderer, clazz);
return Property.as((Property<T>) input.getDefaults().skipCache(), renderer, clazz);
}
@SuppressWarnings("unchecked")
private static <T> Object resolveDefaultPropertyAsList(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
return Property.asList((Property<List<T>>) input.getDefaults(), renderer, clazz);
return Property.asList((Property<List<T>>) input.getDefaults().skipCache(), renderer, clazz);
}
private RunContext buildRunContextForExecutionAndInputs(final FlowInterface flow, final Execution execution, Map<String, InputAndValue> dependencies, final boolean decryptSecrets) {
@@ -502,8 +498,8 @@ public class FlowInputOutput {
yield storageInterface.from(execution, id, current.toString().substring(current.toString().lastIndexOf("/") + 1), new File(current.toString()));
}
}
case JSON -> JacksonMapper.toObject(current.toString());
case YAML -> YAML_MAPPER.readValue(current.toString(), JacksonMapper.OBJECT_TYPE_REFERENCE);
case JSON -> (current instanceof Map || current instanceof Collection<?>) ? current : JacksonMapper.toObject(current.toString());
case YAML -> (current instanceof Map || current instanceof Collection<?>) ? current : YAML_MAPPER.readValue(current.toString(), JacksonMapper.OBJECT_TYPE_REFERENCE);
case URI -> {
Matcher matcher = URI_PATTERN.matcher(current.toString());
if (matcher.matches()) {

View File

@@ -11,6 +11,7 @@ import io.kestra.core.models.flows.State;
import io.kestra.core.models.tasks.ResolvedTask;
import io.kestra.core.models.tasks.Task;
import io.kestra.core.serializers.JacksonMapper;
import io.kestra.core.utils.ListUtils;
import io.kestra.plugin.core.flow.Dag;
import java.util.*;
@@ -143,6 +144,13 @@ public class FlowableUtils {
return Collections.emptyList();
}
// have submitted, leave
Optional<TaskRun> lastSubmitted = execution.findLastSubmitted(taskRuns);
if (lastSubmitted.isPresent()) {
return Collections.emptyList();
}
// last success, find next
Optional<TaskRun> lastTerminated = execution.findLastTerminated(taskRuns);
if (lastTerminated.isPresent()) {
@@ -150,14 +158,41 @@ public class FlowableUtils {
if (currentTasks.size() > lastIndex + 1) {
return Collections.singletonList(currentTasks.get(lastIndex + 1).toNextTaskRunIncrementIteration(execution, parentTaskRun.getIteration()));
} else {
return Collections.singletonList(currentTasks.getFirst().toNextTaskRunIncrementIteration(execution, parentTaskRun.getIteration()));
}
}
return Collections.emptyList();
}
public static Optional<State.Type> resolveSequentialState(
Execution execution,
List<ResolvedTask> tasks,
List<ResolvedTask> errors,
List<ResolvedTask> _finally,
TaskRun parentTaskRun,
RunContext runContext,
boolean allowFailure,
boolean allowWarning
) {
if (ListUtils.emptyOnNull(tasks).stream()
.filter(resolvedTask -> !resolvedTask.getTask().getDisabled())
.findAny()
.isEmpty()) {
return Optional.of(State.Type.SUCCESS);
}
return resolveState(
execution,
tasks,
errors,
_finally,
parentTaskRun,
runContext,
allowFailure,
allowWarning
);
}
public static Optional<State.Type> resolveState(
Execution execution,
List<ResolvedTask> tasks,
@@ -213,7 +248,7 @@ public class FlowableUtils {
}
} else {
// first call, the error flow is not ready, we need to notify the parent task that can be failed to init error flows
if (execution.hasFailed(tasks, parentTaskRun) || terminalState == State.Type.FAILED) {
if (execution.hasFailedNoRetry(tasks, parentTaskRun) || terminalState == State.Type.FAILED) {
return Optional.of(execution.guessFinalState(tasks, parentTaskRun, allowFailure, allowWarning, terminalState));
}
}

View File

@@ -192,5 +192,16 @@ public abstract class RunContext implements PropertyContext {
public record FlowInfo(String tenantId, String namespace, String id, Integer revision) {
}
/**
* @deprecated there is no legitimate use case of this method outside the run context internal self-usage, so it should not be part of the interface
*/
@Deprecated(since = "1.2.0", forRemoval = true)
public abstract boolean isInitialized();
/**
* Get access to the ACL checker.
* Plugins are responsible for using the ACL checker when they access restricted resources, for example,
* when Namespace ACLs are used (EE).
*/
public abstract AclChecker acl();
}

View File

@@ -12,9 +12,10 @@ import io.kestra.core.models.property.PropertyContext;
import io.kestra.core.models.tasks.Task;
import io.kestra.core.models.triggers.AbstractTrigger;
import io.kestra.core.plugins.PluginConfigurations;
import io.kestra.core.services.FlowService;
import io.kestra.core.services.KVStoreService;
import io.kestra.core.services.NamespaceService;
import io.kestra.core.storages.InternalStorage;
import io.kestra.core.storages.NamespaceFactory;
import io.kestra.core.storages.StorageContext;
import io.kestra.core.storages.StorageInterface;
import io.micronaut.context.ApplicationContext;
@@ -48,7 +49,7 @@ public class RunContextFactory {
protected StorageInterface storageInterface;
@Inject
protected FlowService flowService;
protected NamespaceService namespaceService;
@Inject
protected MetricRegistry metricRegistry;
@@ -76,6 +77,9 @@ public class RunContextFactory {
@Inject
private KVStoreService kvStoreService;
@Inject
private NamespaceFactory namespaceFactory;
// hacky
public RunContextInitializer initializer() {
return applicationContext.getBean(RunContextInitializer.class);
@@ -103,7 +107,7 @@ public class RunContextFactory {
.withLogger(runContextLogger)
// Execution
.withPluginConfiguration(Map.of())
.withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forExecution(execution), storageInterface, flowService))
.withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forExecution(execution), storageInterface, namespaceService, namespaceFactory))
.withVariableRenderer(variableRenderer)
.withVariables(runVariableModifier.apply(
newRunVariablesBuilder()
@@ -133,7 +137,7 @@ public class RunContextFactory {
.withLogger(runContextLogger)
// Task
.withPluginConfiguration(pluginConfigurations.getConfigurationByPluginTypeOrAliases(task.getType(), task.getClass()))
.withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, flowService))
.withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, namespaceService, namespaceFactory))
.withVariables(newRunVariablesBuilder()
.withFlow(flow)
.withTask(task)
@@ -167,14 +171,16 @@ public class RunContextFactory {
.build();
}
@VisibleForTesting
public RunContext of(final FlowInterface flow, final Map<String, Object> variables) {
RunContextLogger runContextLogger = new RunContextLogger();
return newBuilder()
.withLogger(runContextLogger)
.withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forFlow(flow), storageInterface, flowService))
.withVariables(variables)
.withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forFlow(flow), storageInterface, namespaceService, namespaceFactory))
.withVariables(newRunVariablesBuilder()
.withFlow(flow)
.withVariables(variables)
.build(runContextLogger, PropertyContext.create(this.variableRenderer))
)
.withSecretInputs(secretInputsFromFlow(flow))
.build();
}
@@ -212,7 +218,8 @@ public class RunContextFactory {
}
},
storageInterface,
flowService
namespaceService,
namespaceFactory
))
.withVariables(variables)
.withTask(task)

View File

@@ -8,8 +8,9 @@ import io.kestra.core.models.tasks.runners.TaskRunner;
import io.kestra.core.models.triggers.AbstractTrigger;
import io.kestra.core.models.triggers.TriggerContext;
import io.kestra.core.plugins.PluginConfigurations;
import io.kestra.core.services.FlowService;
import io.kestra.core.services.NamespaceService;
import io.kestra.core.storages.InternalStorage;
import io.kestra.core.storages.NamespaceFactory;
import io.kestra.core.storages.StorageContext;
import io.kestra.core.storages.StorageInterface;
import io.kestra.core.utils.IdUtils;
@@ -44,7 +45,10 @@ public class RunContextInitializer {
protected StorageInterface storageInterface;
@Inject
protected FlowService flowService;
protected NamespaceFactory namespaceFactory;
@Inject
protected NamespaceService namespaceService;
@Value("${kestra.encryption.secret-key}")
protected Optional<String> secretKey;
@@ -135,7 +139,7 @@ public class RunContextInitializer {
runContext.setVariables(enrichedVariables);
runContext.setPluginConfiguration(pluginConfigurations.getConfigurationByPluginTypeOrAliases(task.getType(), task.getClass()));
runContext.setStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, flowService));
runContext.setStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forTask(taskRun), storageInterface, namespaceService, namespaceFactory));
runContext.setLogger(runContextLogger);
runContext.setTask(task);
@@ -230,7 +234,8 @@ public class RunContextInitializer {
runContextLogger.logger(),
context,
storageInterface,
flowService
namespaceService,
namespaceFactory
);
runContext.setLogger(runContextLogger);

View File

@@ -2,6 +2,7 @@ package io.kestra.core.runners.pebble;
import io.kestra.core.runners.VariableRenderer;
import io.kestra.core.runners.pebble.functions.RenderingFunctionInterface;
import io.micrometer.core.instrument.MeterRegistry;
import io.micronaut.context.ApplicationContext;
import io.micronaut.core.annotation.Nullable;
import io.pebbletemplates.pebble.PebbleEngine;
@@ -18,35 +19,37 @@ import java.util.stream.Collectors;
@Singleton
public class PebbleEngineFactory {
private final ApplicationContext applicationContext;
private final VariableRenderer.VariableConfiguration variableConfiguration;
private final MeterRegistry meterRegistry;
@Inject
public PebbleEngineFactory(ApplicationContext applicationContext, @Nullable VariableRenderer.VariableConfiguration variableConfiguration) {
public PebbleEngineFactory(ApplicationContext applicationContext, @Nullable VariableRenderer.VariableConfiguration variableConfiguration, MeterRegistry meterRegistry) {
this.applicationContext = applicationContext;
this.variableConfiguration = variableConfiguration;
this.meterRegistry = meterRegistry;
}
public PebbleEngine create() {
PebbleEngine.Builder builder = newPebbleEngineBuilder();
this.applicationContext.getBeansOfType(Extension.class).forEach(builder::extension);
return builder.build();
}
public PebbleEngine createWithMaskedFunctions(VariableRenderer renderer, final List<String> functionsToMask) {
PebbleEngine.Builder builder = newPebbleEngineBuilder();
this.applicationContext.getBeansOfType(Extension.class).stream()
.map(e -> functionsToMask.stream().anyMatch(fun -> e.getFunctions().containsKey(fun))
? extensionWithMaskedFunctions(renderer, e, functionsToMask)
: e)
.forEach(builder::extension);
return builder.build();
}
private PebbleEngine.Builder newPebbleEngineBuilder() {
PebbleEngine.Builder builder = new PebbleEngine.Builder()
.registerExtensionCustomizer(ExtensionCustomizer::new)
@@ -54,13 +57,15 @@ public class PebbleEngineFactory {
.cacheActive(this.variableConfiguration.getCacheEnabled())
.newLineTrimming(false)
.autoEscaping(false);
if (this.variableConfiguration.getCacheEnabled()) {
builder = builder.templateCache(new PebbleLruCache(this.variableConfiguration.getCacheSize()));
PebbleLruCache cache = new PebbleLruCache(this.variableConfiguration.getCacheSize());
cache.register(meterRegistry);
builder = builder.templateCache(cache);
}
return builder;
}
private Extension extensionWithMaskedFunctions(VariableRenderer renderer, Extension initialExtension, List<String> maskedFunctions) {
return (Extension) Proxy.newProxyInstance(
initialExtension.getClass().getClassLoader(),
@@ -74,16 +79,16 @@ public class PebbleEngineFactory {
} else if (RenderingFunctionInterface.class.isAssignableFrom(entry.getValue().getClass())) {
return Map.entry(entry.getKey(), this.variableRendererProxy(renderer, entry.getValue()));
}
return entry;
}).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
return method.invoke(initialExtension, methodArgs);
}
);
}
private Function variableRendererProxy(VariableRenderer renderer, Function initialFunction) {
return (Function) Proxy.newProxyInstance(
initialFunction.getClass().getClassLoader(),
@@ -96,7 +101,7 @@ public class PebbleEngineFactory {
}
);
}
private Function maskedFunctionProxy(Function initialFunction) {
return (Function) Proxy.newProxyInstance(
initialFunction.getClass().getClassLoader(),

View File

@@ -1,29 +1,29 @@
package io.kestra.core.runners.pebble;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.binder.cache.CaffeineCacheMetrics;
import io.pebbletemplates.pebble.cache.PebbleCache;
import io.pebbletemplates.pebble.template.PebbleTemplate;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
@Slf4j
public class PebbleLruCache implements PebbleCache<Object, PebbleTemplate> {
Cache<Object, PebbleTemplate> cache;
private final Cache<Object, PebbleTemplate> cache;
public PebbleLruCache(int maximumSize) {
cache = CacheBuilder.newBuilder()
cache = Caffeine.newBuilder()
.initialCapacity(250)
.maximumSize(maximumSize)
.recordStats()
.build();
}
@Override
public PebbleTemplate computeIfAbsent(Object key, Function<? super Object, ? extends PebbleTemplate> mappingFunction) {
try {
return cache.get(key, () -> mappingFunction.apply(key));
return cache.get(key, mappingFunction);
} catch (Exception e) {
// we retry the mapping function in order to let the exception be thrown instead of being capture by cache
return mappingFunction.apply(key);
@@ -34,4 +34,8 @@ public class PebbleLruCache implements PebbleCache<Object, PebbleTemplate> {
public void invalidateAll() {
cache.invalidateAll();
}
public void register(MeterRegistry meterRegistry) {
CaffeineCacheMetrics.monitor(meterRegistry, cache, "pebble-template");
}
}

View File

@@ -2,11 +2,8 @@ package io.kestra.core.runners.pebble.functions;
import io.kestra.core.runners.LocalPath;
import io.kestra.core.runners.LocalPathFactory;
import io.kestra.core.services.FlowService;
import io.kestra.core.storages.InternalNamespace;
import io.kestra.core.storages.Namespace;
import io.kestra.core.storages.StorageContext;
import io.kestra.core.storages.StorageInterface;
import io.kestra.core.services.NamespaceService;
import io.kestra.core.storages.*;
import io.kestra.core.utils.Slugify;
import io.micronaut.context.annotation.Value;
import io.pebbletemplates.pebble.error.PebbleException;
@@ -36,7 +33,7 @@ abstract class AbstractFileFunction implements Function {
private static final Pattern EXECUTION_FILE = Pattern.compile(".*/.*/executions/.*/tasks/.*/.*");
@Inject
protected FlowService flowService;
protected NamespaceService namespaceService;
@Inject
protected StorageInterface storageInterface;
@@ -44,6 +41,9 @@ abstract class AbstractFileFunction implements Function {
@Inject
protected LocalPathFactory localPathFactory;
@Inject
protected NamespaceFactory namespaceFactory;
@Value("${" + LocalPath.ENABLE_FILE_FUNCTIONS_CONFIG + ":true}")
protected boolean enableFileProtocol;
@@ -81,23 +81,21 @@ abstract class AbstractFileFunction implements Function {
} else if (str.startsWith(LocalPath.FILE_PROTOCOL)) {
fileUri = URI.create(str);
namespace = checkEnabledLocalFileAndReturnNamespace(args, flow);
} else if(str.startsWith(Namespace.NAMESPACE_FILE_SCHEME)) {
URI nsFileUri = URI.create(str);
namespace = checkedAllowedNamespaceAndReturnNamespace(args, nsFileUri, tenantId, flow);
InternalNamespace internalNamespace = new InternalNamespace(flow.get(TENANT_ID), namespace, storageInterface);
fileUri = internalNamespace.get(Path.of(nsFileUri.getPath())).uri();
} else if (str.startsWith(Namespace.NAMESPACE_FILE_SCHEME)) {
fileUri = URI.create(str);
namespace = checkedAllowedNamespaceAndReturnNamespace(args, fileUri, tenantId, flow);
} else if (URI_PATTERN.matcher(str).matches()) {
// it is an unsupported URI
throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(str));
} else {
fileUri = URI.create(Namespace.NAMESPACE_FILE_SCHEME + ":///" + str);
namespace = (String) Optional.ofNullable(args.get(NAMESPACE)).orElse(flow.get(NAMESPACE));
fileUri = URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.namespaceFilePrefix(namespace) + "/" + str);
flowService.checkAllowedNamespace(tenantId, namespace, tenantId, flow.get(NAMESPACE));
namespaceService.checkAllowedNamespace(tenantId, namespace, tenantId, flow.get(NAMESPACE));
}
} else {
throw new PebbleException(null, "Unable to read the file " + path, lineNumber, self.getName());
}
return fileFunction(context, fileUri, namespace, tenantId);
return fileFunction(context, fileUri, namespace, tenantId, args);
} catch (IOException e) {
throw new PebbleException(e, e.getMessage(), lineNumber, self.getName());
}
@@ -110,7 +108,7 @@ abstract class AbstractFileFunction implements Function {
protected abstract String getErrorMessage();
protected abstract Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException;
protected abstract Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException;
boolean isFileUriValid(String namespace, String flowId, String executionId, URI path) {
// Internal storage URI should be: kestra:///$namespace/$flowId/executions/$executionId/tasks/$taskName/$taskRunId/$random.ion or kestra:///$namespace/$flowId/executions/$executionId/trigger/$triggerName/$random.ion
@@ -177,7 +175,7 @@ abstract class AbstractFileFunction implements Function {
// 5. replace '/' with '.'
namespace = namespace.replace("/", ".");
flowService.checkAllowedNamespace(tenantId, namespace, tenantId, fromNamespace);
namespaceService.checkAllowedNamespace(tenantId, namespace, tenantId, fromNamespace);
return namespace;
}
@@ -198,7 +196,7 @@ abstract class AbstractFileFunction implements Function {
// we will transform nsfile URI into a kestra URI so it is handled seamlessly by all functions
String customNs = Optional.ofNullable((String) args.get(NAMESPACE)).orElse(nsFileUri.getAuthority());
if (customNs != null) {
flowService.checkAllowedNamespace(tenantId, customNs, tenantId, flow.get(NAMESPACE));
namespaceService.checkAllowedNamespace(tenantId, customNs, tenantId, flow.get(NAMESPACE));
}
return Optional.ofNullable(customNs).orElse(flow.get(NAMESPACE));
}

View File

@@ -3,7 +3,7 @@ package io.kestra.core.runners.pebble.functions;
import io.kestra.core.models.executions.LogEntry;
import io.kestra.core.models.tasks.retrys.Exponential;
import io.kestra.core.runners.pebble.PebbleUtils;
import io.kestra.core.services.LogService;
import io.kestra.core.services.ExecutionLogService;
import io.kestra.core.utils.ListUtils;
import io.kestra.core.utils.RetryUtils;
import io.micronaut.context.annotation.Requires;
@@ -23,14 +23,11 @@ import java.util.Map;
@Requires(property = "kestra.repository.type")
public class ErrorLogsFunction implements Function {
@Inject
private LogService logService;
private ExecutionLogService logService;
@Inject
private PebbleUtils pebbleUtils;
@Inject
private RetryUtils retryUtils;
@Override
public List<String> getArgumentNames() {
return Collections.emptyList();
@@ -46,7 +43,7 @@ public class ErrorLogsFunction implements Function {
Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
Map<String, String> execution = (Map<String, String>) context.getVariable("execution");
RetryUtils.Instance<List<LogEntry>, Throwable> retry = retryUtils.of(Exponential.builder()
RetryUtils.Instance<List<LogEntry>, Throwable> retry = RetryUtils.of(Exponential.builder()
.delayFactor(2.0)
.interval(Duration.ofMillis(100))
.maxInterval(Duration.ofSeconds(1))

View File

@@ -1,22 +1,30 @@
package io.kestra.core.runners.pebble.functions;
import io.kestra.core.runners.LocalPath;
import io.kestra.core.storages.Namespace;
import io.kestra.core.storages.NamespaceFile;
import io.kestra.core.storages.StorageContext;
import io.pebbletemplates.pebble.template.EvaluationContext;
import jakarta.inject.Singleton;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Path;
import java.util.Map;
@Singleton
public class FileExistsFunction extends AbstractFileFunction {
private static final String ERROR_MESSAGE = "The 'fileExists' function expects an argument 'path' that is a path to the internal storage URI.";
@Override
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
return switch (path.getScheme()) {
case StorageContext.KESTRA_SCHEME -> storageInterface.exists(tenantId, namespace, path);
case LocalPath.FILE_SCHEME -> localPathFactory.createLocalPath().exists(path);
case Namespace.NAMESPACE_FILE_SCHEME -> {
Namespace namespaceStorage = namespaceFactory.of(tenantId, namespace, storageInterface);
yield namespaceStorage.exists(NamespaceFile.normalize(Path.of(path.getPath()), true));
}
default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
};
}

View File

@@ -2,19 +2,23 @@ package io.kestra.core.runners.pebble.functions;
import io.kestra.core.runners.LocalPath;
import io.kestra.core.storages.FileAttributes;
import io.kestra.core.storages.Namespace;
import io.kestra.core.storages.NamespaceFile;
import io.kestra.core.storages.StorageContext;
import io.pebbletemplates.pebble.template.EvaluationContext;
import jakarta.inject.Singleton;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Path;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Map;
@Singleton
public class FileSizeFunction extends AbstractFileFunction {
private static final String ERROR_MESSAGE = "The 'fileSize' function expects an argument 'path' that is a path to the internal storage URI.";
@Override
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
return switch (path.getScheme()) {
case StorageContext.KESTRA_SCHEME -> {
FileAttributes fileAttributes = storageInterface.getAttributes(tenantId, namespace, path);
@@ -24,6 +28,12 @@ public class FileSizeFunction extends AbstractFileFunction {
BasicFileAttributes fileAttributes = localPathFactory.createLocalPath().getAttributes(path);
yield fileAttributes.size();
}
case Namespace.NAMESPACE_FILE_SCHEME -> {
FileAttributes fileAttributes = namespaceFactory
.of(tenantId, namespace, storageInterface)
.getFileMetadata(NamespaceFile.normalize(Path.of(path.getPath()), true));
yield fileAttributes.getSize();
}
default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
};
}

View File

@@ -1,19 +1,24 @@
package io.kestra.core.runners.pebble.functions;
import io.kestra.core.runners.LocalPath;
import io.kestra.core.storages.FileAttributes;
import io.kestra.core.storages.Namespace;
import io.kestra.core.storages.NamespaceFile;
import io.kestra.core.storages.StorageContext;
import io.pebbletemplates.pebble.template.EvaluationContext;
import jakarta.inject.Singleton;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.nio.file.Path;
import java.util.Map;
@Singleton
public class IsFileEmptyFunction extends AbstractFileFunction {
private static final String ERROR_MESSAGE = "The 'isFileEmpty' function expects an argument 'path' that is a path to a namespace file or an internal storage URI.";
@Override
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
return switch (path.getScheme()) {
case StorageContext.KESTRA_SCHEME -> {
try (InputStream inputStream = storageInterface.get(tenantId, namespace, path)) {
@@ -27,6 +32,12 @@ public class IsFileEmptyFunction extends AbstractFileFunction {
yield inputStream.read(buffer, 0, 1) <= 0;
}
}
case Namespace.NAMESPACE_FILE_SCHEME -> {
FileAttributes fileAttributes = namespaceFactory
.of(tenantId, namespace, storageInterface)
.getFileMetadata(NamespaceFile.normalize(Path.of(path.getPath()), true));
yield fileAttributes.getSize() <= 0;
}
default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
};
}
@@ -35,4 +46,4 @@ public class IsFileEmptyFunction extends AbstractFileFunction {
protected String getErrorMessage() {
return ERROR_MESSAGE;
}
}
}

View File

@@ -1,20 +1,37 @@
package io.kestra.core.runners.pebble.functions;
import io.kestra.core.runners.LocalPath;
import io.kestra.core.storages.Namespace;
import io.kestra.core.storages.NamespaceFile;
import io.kestra.core.storages.StorageContext;
import io.pebbletemplates.pebble.template.EvaluationContext;
import jakarta.inject.Singleton;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
@Singleton
public class ReadFileFunction extends AbstractFileFunction {
public static final String VERSION = "version";
private static final String ERROR_MESSAGE = "The 'read' function expects an argument 'path' that is a path to a namespace file or an internal storage URI.";
@Override
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId) throws IOException {
public List<String> getArgumentNames() {
return Stream.concat(
super.getArgumentNames().stream(),
Stream.of(VERSION)
).toList();
}
@Override
protected Object fileFunction(EvaluationContext context, URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
return switch (path.getScheme()) {
case StorageContext.KESTRA_SCHEME -> {
try (InputStream inputStream = storageInterface.get(tenantId, namespace, path)) {
@@ -26,12 +43,30 @@ public class ReadFileFunction extends AbstractFileFunction {
yield new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
}
}
case Namespace.NAMESPACE_FILE_SCHEME -> {
try (InputStream inputStream = contentInputStream(path, namespace, tenantId, args)) {
yield new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
}
}
default -> throw new IllegalArgumentException(SCHEME_NOT_SUPPORTED_ERROR.formatted(path));
};
}
private InputStream contentInputStream(URI path, String namespace, String tenantId, Map<String, Object> args) throws IOException {
Namespace namespaceStorage = namespaceFactory.of(tenantId, namespace, storageInterface);
if (args.containsKey(VERSION)) {
return namespaceStorage.getFileContent(
NamespaceFile.normalize(Path.of(path.getPath()), true),
Integer.parseInt(args.get(VERSION).toString())
);
}
return namespaceStorage.getFileContent(NamespaceFile.normalize(Path.of(path.getPath()), true));
}
@Override
protected String getErrorMessage() {
return ERROR_MESSAGE;
}
}
}

View File

@@ -9,6 +9,7 @@ import io.kestra.core.secret.SecretNotFoundException;
import io.kestra.core.secret.SecretService;
import io.kestra.core.serializers.JacksonMapper;
import io.kestra.core.services.FlowService;
import io.kestra.core.services.NamespaceService;
import io.pebbletemplates.pebble.error.PebbleException;
import io.pebbletemplates.pebble.extension.Function;
import io.pebbletemplates.pebble.template.EvaluationContext;
@@ -36,7 +37,7 @@ public class SecretFunction implements Function {
private SecretService secretService;
@Inject
private FlowService flowService;
private NamespaceService namespaceService;
@Override
public List<String> getArgumentNames() {
@@ -56,7 +57,7 @@ public class SecretFunction implements Function {
if (namespace == null) {
namespace = flowNamespace;
} else {
flowService.checkAllowedNamespace(flowTenantId, namespace, flowTenantId, flowNamespace);
namespaceService.checkAllowedNamespace(flowTenantId, namespace, flowTenantId, flowNamespace);
}
try {

View File

@@ -26,7 +26,14 @@ public class ListOrMapOfLabelDeserializer extends JsonDeserializer<List<Label>>
else if (p.hasToken(JsonToken.START_ARRAY)) {
// deserialize as list
List<Map<String, String>> ret = ctxt.readValue(p, List.class);
return ret.stream().map(map -> new Label(map.get("key"), map.get("value"))).toList();
return ret.stream().map(map -> {
Object value = map.get("value");
if (isAllowedType(value)) {
return new Label(map.get("key"), String.valueOf(value));
} else {
throw new IllegalArgumentException("Unsupported type for key: " + map.get("key") + ", value: " + value);
}
}).toList();
}
else if (p.hasToken(JsonToken.START_OBJECT)) {
// deserialize as map

View File

@@ -2,12 +2,15 @@ package io.kestra.core.services;
import io.kestra.core.models.executions.LogEntry;
import io.kestra.core.repositories.LogRepositoryInterface;
import io.micronaut.data.model.Pageable;
import io.micronaut.data.model.Sort;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
import org.slf4j.event.Level;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.time.ZonedDateTime;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -17,9 +20,42 @@ import java.util.stream.Stream;
*/
@Singleton
public class ExecutionLogService {
private final LogRepositoryInterface logRepository;
@Inject
private LogRepositoryInterface logRepository;
public ExecutionLogService(LogRepositoryInterface logRepository) {
this.logRepository = logRepository;
}
/**
* Purges log entries matching the given criteria.
*
* @param tenantId the tenant identifier
* @param namespace the namespace of the flow
* @param flowId the flow identifier
* @param executionId the execution identifier
* @param logLevels the list of log levels to delete
* @param startDate the start of the date range
* @param endDate the end of the date range.
* @return the number of log entries deleted
*/
public int purge(String tenantId, String namespace, String flowId, String executionId, List<Level> logLevels, ZonedDateTime startDate, ZonedDateTime endDate) {
return logRepository.deleteByQuery(tenantId, namespace, flowId, executionId, logLevels, startDate, endDate);
}
/**
* Fetches the error logs of an execution.
* <p>
* This method limits the results to the first 25 error logs, ordered by timestamp asc.
*
* @return the log entries
*/
public List<LogEntry> errorLogs(String tenantId, String executionId) {
return logRepository.findByExecutionId(tenantId, executionId, Level.ERROR, Pageable.from(1, 25, Sort.of(Sort.Order.asc("timestamp"))));
}
public InputStream getExecutionLogsAsStream(String tenantId,
String executionId,
Level minLevel,

View File

@@ -2,8 +2,10 @@ package io.kestra.core.services;
import com.fasterxml.jackson.core.JsonProcessingException;
import io.kestra.core.exceptions.FlowProcessingException;
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
import io.kestra.core.models.executions.Execution;
import io.kestra.core.models.flows.*;
import io.kestra.core.models.flows.check.Check;
import io.kestra.core.models.tasks.RunnableTask;
import io.kestra.core.models.topologies.FlowTopology;
import io.kestra.core.models.triggers.AbstractTrigger;
@@ -12,10 +14,13 @@ import io.kestra.core.models.validations.ValidateConstraintViolation;
import io.kestra.core.plugins.PluginRegistry;
import io.kestra.core.repositories.FlowRepositoryInterface;
import io.kestra.core.repositories.FlowTopologyRepositoryInterface;
import io.kestra.core.runners.RunContext;
import io.kestra.core.runners.RunContextFactory;
import io.kestra.core.serializers.JacksonMapper;
import io.kestra.core.utils.ListUtils;
import io.kestra.plugin.core.flow.Pause;
import jakarta.inject.Inject;
import jakarta.inject.Provider;
import jakarta.inject.Singleton;
import jakarta.validation.ConstraintViolationException;
import lombok.extern.slf4j.Slf4j;
@@ -54,6 +59,9 @@ public class FlowService {
@Inject
Optional<FlowTopologyRepositoryInterface> flowTopologyRepository;
@Inject
Provider<RunContextFactory> runContextFactory; // Lazy init: avoid circular dependency error.
/**
* Validates and creates the given flow.
* <p>
@@ -85,6 +93,50 @@ public class FlowService {
.orElseThrow(() -> new IllegalStateException("Cannot perform operation on flow. Cause: No FlowRepository"));
}
/**
* Evaluates all checks defined in the given flow using the provided inputs.
* <p>
* Each check's {@link Check#getCondition()} is evaluated in the context of the flow.
* If a condition evaluates to {@code false} or fails to evaluate due to a
* variable error, the corresponding {@link Check} is added to the returned list.
* </p>
*
* @param flow the flow containing the checks to evaluate
* @param inputs the input values used when evaluating the conditions
* @return a list of checks whose conditions evaluated to {@code false} or failed to evaluate
*/
public List<Check> getFailedChecks(Flow flow, Map<String, Object> inputs) {
if (!ListUtils.isEmpty(flow.getChecks())) {
RunContext runContext = runContextFactory.get().of(flow, Map.of("inputs", inputs));
List<Check> falseConditions = new ArrayList<>();
for (Check check : flow.getChecks()) {
try {
boolean result = Boolean.TRUE.equals(runContext.renderTyped(check.getCondition()));
if (!result) {
falseConditions.add(check);
}
} catch (IllegalVariableEvaluationException e) {
log.debug("[tenant: {}] [namespace: {}] [flow: {}] Failed to evaluate check condition. Cause.: {}",
flow.getTenantId(),
flow.getNamespace(),
flow.getId(),
e.getMessage(),
e
);
falseConditions.add(Check
.builder()
.message("Failed to evaluate check condition. Cause: " + e.getMessage())
.behavior(Check.Behavior.BLOCK_EXECUTION)
.style(Check.Style.ERROR)
.build()
);
}
}
return falseConditions;
}
return List.of();
}
/**
* Validates the given flow source.
* <p>
@@ -456,50 +508,6 @@ public class FlowService {
return flowRepository.get().delete(flow);
}
/**
* Return true if the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
* As namespace restriction is an EE feature, this will always return true in OSS.
*/
public boolean isAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
return true;
}
/**
* Check that the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
* If not, throw an IllegalArgumentException.
*/
public void checkAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
if (!isAllowedNamespace(tenant, namespace, fromTenant, fromNamespace)) {
throw new IllegalArgumentException("Namespace " + namespace + " is not allowed.");
}
}
/**
* Return true if the namespace is allowed from all the namespace in the 'fromTenant' tenant.
* As namespace restriction is an EE feature, this will always return true in OSS.
*/
public boolean areAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
return true;
}
/**
* Check that the namespace is allowed from all the namespace in the 'fromTenant' tenant.
* If not, throw an IllegalArgumentException.
*/
public void checkAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
if (!areAllowedAllNamespaces(tenant, fromTenant, fromNamespace)) {
throw new IllegalArgumentException("All namespaces are not allowed, you should either filter on a namespace or configure all namespaces to allow your namespace.");
}
}
/**
* Return true if require existing namespace is enabled and the namespace didn't already exist.
* As namespace management is an EE feature, this will always return false in OSS.
*/
public boolean requireExistingNamespace(String tenant, String namespace) {
return false;
}
/**
* Gets the executable flow for the given namespace, id, and revision.
* Warning: this method bypasses ACL so someone with only execution right can create a flow execution

View File

@@ -20,9 +20,6 @@ public class KVStoreService {
@Inject
private StorageInterface storageInterface;
@Inject
private FlowService flowService;
@Inject
private NamespaceService namespaceService;
@@ -38,7 +35,7 @@ public class KVStoreService {
boolean isNotSameNamespace = fromNamespace != null && !namespace.equals(fromNamespace);
if (isNotSameNamespace && isNotParentNamespace(namespace, fromNamespace)) {
try {
flowService.checkAllowedNamespace(tenant, namespace, tenant, fromNamespace);
namespaceService.checkAllowedNamespace(tenant, namespace, tenant, fromNamespace);
} catch (IllegalArgumentException e) {
throw new KVStoreException(String.format(
"Cannot access the KV store. Access to '%s' namespace is not allowed from '%s'.", namespace, fromNamespace)

View File

@@ -1,5 +1,6 @@
package io.kestra.core.services;
import io.kestra.core.exceptions.ResourceAccessDeniedException;
import io.kestra.core.repositories.FlowRepositoryInterface;
import io.kestra.core.utils.NamespaceUtils;
import jakarta.inject.Inject;
@@ -39,4 +40,52 @@ public class NamespaceService {
}
return false;
}
/**
* Return true if require existing namespace is enabled and the namespace didn't already exist.
* As namespace management is an EE feature, this will always return false in OSS.
*/
public boolean requireExistingNamespace(String tenant, String namespace) {
return false;
}
/**
* Return true if the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
* As namespace restriction is an EE feature, this will always return true in OSS.
*/
public boolean isAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
return true;
}
/**
* Check that the namespace is allowed from the namespace denoted by 'fromTenant' and 'fromNamespace'.
* If not, throw a ResourceAccessDeniedException.
*
* @throws ResourceAccessDeniedException if the namespace is not allowed.
*/
public void checkAllowedNamespace(String tenant, String namespace, String fromTenant, String fromNamespace) {
if (!isAllowedNamespace(tenant, namespace, fromTenant, fromNamespace)) {
throw new ResourceAccessDeniedException("Namespace " + namespace + " is not allowed.");
}
}
/**
* Return true if the namespace is allowed from all the namespace in the 'fromTenant' tenant.
* As namespace restriction is an EE feature, this will always return true in OSS.
*/
public boolean areAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
return true;
}
/**
* Check that the namespace is allowed from all the namespace in the 'fromTenant' tenant.
* If not, throw a ResourceAccessDeniedException.
*
* @throws ResourceAccessDeniedException if all namespaces all aren't allowed.
*/
public void checkAllowedAllNamespaces(String tenant, String fromTenant, String fromNamespace) {
if (!areAllowedAllNamespaces(tenant, fromTenant, fromNamespace)) {
throw new ResourceAccessDeniedException("All namespaces are not allowed, you should either filter on a namespace or configure all namespaces to allow your namespace.");
}
}
}

View File

@@ -23,6 +23,7 @@ import io.kestra.core.queues.QueueInterface;
import io.kestra.core.runners.RunContextLogger;
import io.kestra.core.serializers.JacksonMapper;
import io.kestra.core.serializers.YamlParser;
import io.kestra.core.utils.Logs;
import io.kestra.core.utils.MapUtils;
import io.kestra.plugin.core.flow.Template;
import io.micronaut.context.annotation.Value;
@@ -30,7 +31,6 @@ import io.micronaut.core.annotation.Nullable;
import jakarta.annotation.PostConstruct;
import jakarta.inject.Inject;
import jakarta.inject.Named;
import jakarta.inject.Provider;
import jakarta.inject.Singleton;
import jakarta.validation.ConstraintViolationException;
import lombok.extern.slf4j.Slf4j;
@@ -82,10 +82,7 @@ public class PluginDefaultService {
@Inject
protected PluginRegistry pluginRegistry;
@Inject
protected Provider<LogService> logService; // lazy-init
@Value("{kestra.templates.enabled:false}")
private boolean templatesEnabled;
@@ -255,7 +252,7 @@ public class PluginDefaultService {
if (source == null) {
// This should never happen
String error = "Cannot apply plugin defaults. Cause: flow has no defined source.";
logService.get().logExecution(flow, log, Level.ERROR, error);
Logs.logExecution(flow, log, Level.ERROR, error);
throw new IllegalArgumentException(error);
}
@@ -311,7 +308,7 @@ public class PluginDefaultService {
result = parseFlowWithAllDefaults(flow.getTenantId(), flow.getNamespace(), flow.getRevision(), flow.isDeleted(), source, true, false);
} catch (Exception e) {
if (safe) {
logService.get().logExecution(flow, log, Level.ERROR, "Failed to read flow.", e);
Logs.logExecution(flow, log, Level.ERROR, "Failed to read flow.", e);
result = FlowWithException.from(flow, e);
// deleted is not part of the original 'source'

View File

@@ -1,18 +1,27 @@
package io.kestra.core.storages;
import io.kestra.core.models.FetchVersion;
import io.kestra.core.models.QueryFilter;
import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
import io.kestra.core.repositories.ArrayListTotal;
import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
import io.micronaut.data.model.Pageable;
import jakarta.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Path;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.*;
import java.util.function.Predicate;
import java.util.stream.Stream;
import static io.kestra.core.utils.Rethrow.throwFunction;
/**
* The default {@link Namespace} implementation.
@@ -28,6 +37,7 @@ public class InternalNamespace implements Namespace {
private final String namespace;
private final String tenant;
private final StorageInterface storage;
private final NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository;
private final Logger logger;
/**
@@ -36,8 +46,8 @@ public class InternalNamespace implements Namespace {
* @param namespace The namespace
* @param storage The storage.
*/
public InternalNamespace(@Nullable final String tenant, final String namespace, final StorageInterface storage) {
this(LOG, tenant, namespace, storage);
public InternalNamespace(@Nullable final String tenant, final String namespace, final StorageInterface storage, final NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepository) {
this(LOG, tenant, namespace, storage, namespaceFileMetadataRepository);
}
/**
@@ -48,13 +58,27 @@ public class InternalNamespace implements Namespace {
* @param tenant The tenant.
* @param storage The storage.
*/
public InternalNamespace(final Logger logger, @Nullable final String tenant, final String namespace, final StorageInterface storage) {
public InternalNamespace(final Logger logger, @Nullable final String tenant, final String namespace, final StorageInterface storage, final NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepositoryInterface) {
this.logger = Objects.requireNonNull(logger, "logger cannot be null");
this.namespace = Objects.requireNonNull(namespace, "namespace cannot be null");
this.storage = Objects.requireNonNull(storage, "storage cannot be null");
this.namespaceFileMetadataRepository = Objects.requireNonNull(namespaceFileMetadataRepositoryInterface, "namespaceFileMetadataRepository cannot be null");
this.tenant = tenant;
}
@Override
public ArrayListTotal<NamespaceFile> find(Pageable pageable, List<QueryFilter> filters, boolean allowDeleted, FetchVersion fetchVersion) {
return namespaceFileMetadataRepository.find(
pageable,
tenant,
Stream.concat(filters.stream(), Stream.of(
QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build()
)).toList(),
allowDeleted,
fetchVersion
).map(throwFunction(NamespaceFile::fromMetadata));
}
/**
* {@inheritDoc}
**/
@@ -73,35 +97,106 @@ public class InternalNamespace implements Namespace {
**/
@Override
public List<NamespaceFile> all() throws IOException {
return all(false);
return all(null);
}
/**
* {@inheritDoc}
**/
@Override
public List<NamespaceFile> all(final boolean includeDirectories) throws IOException {
return all(null, includeDirectories);
public List<NamespaceFile> all(final String containing, boolean includeDirectories) throws IOException {
List<NamespaceFileMetadata> namespaceFilesMetadata = namespaceFileMetadataRepository.find(Pageable.UNPAGED, tenant, Stream.concat(
Stream.of(QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build()),
Optional.ofNullable(containing).flatMap(p -> {
if (p.equals("/")) {
return Optional.empty();
}
return Optional.of(QueryFilter.builder().field(QueryFilter.Field.QUERY).operation(QueryFilter.Op.EQUALS).value(p).build());
}).stream()
).toList(), false);
if (!includeDirectories) {
namespaceFilesMetadata = namespaceFilesMetadata.stream().filter(nsFileMetadata -> !nsFileMetadata.isDirectory()).toList();
}
return namespaceFilesMetadata.stream().filter(nsFileMetadata -> !nsFileMetadata.getPath().equals("/")).map(nsFileMetadata -> NamespaceFile.of(namespace, Path.of(nsFileMetadata.getPath()), nsFileMetadata.getVersion())).toList();
}
/**
* {@inheritDoc}
**/
@Override
public List<NamespaceFile> all(final String prefix, final boolean includeDirectories) throws IOException {
URI namespacePrefix = URI.create(NamespaceFile.of(namespace, Optional.ofNullable(prefix).map(Path::of).orElse(null)).storagePath().toString().replace("\\","/") + "/");
return storage.allByPrefix(tenant, namespace, namespacePrefix, includeDirectories)
.stream()
.map(uri -> new NamespaceFile(relativize(uri), uri, namespace))
.toList();
public List<NamespaceFileMetadata> children(String parentPath, boolean recursive) throws IOException {
final String normalizedParentPath = NamespaceFile.normalize(Path.of(parentPath), true).toString();
return namespaceFileMetadataRepository.find(Pageable.UNPAGED, tenant, List.of(
QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
QueryFilter.builder()
.field(QueryFilter.Field.PARENT_PATH)
.operation(recursive ? QueryFilter.Op.STARTS_WITH : QueryFilter.Op.EQUALS)
.value(normalizedParentPath.endsWith("/") ? normalizedParentPath : normalizedParentPath + "/")
.build()
), false);
}
@Override
public List<Pair<NamespaceFile, NamespaceFile>> move(Path source, Path target) throws Exception {
final Path normalizedSource = NamespaceFile.normalize(source, true);
final Path normalizedTarget = NamespaceFile.normalize(target, true);
if (findByPath(normalizedTarget).isPresent()) {
throw new IOException(String.format(
"File '%s' already exists in namespace '%s'.",
normalizedTarget,
namespace
));
}
ArrayListTotal<NamespaceFileMetadata> beforeRename = namespaceFileMetadataRepository.find(Pageable.UNPAGED, tenant, List.of(
QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
QueryFilter.builder().field(QueryFilter.Field.PATH).operation(QueryFilter.Op.IN).value(List.of(normalizedSource.toString(), normalizedSource + "/")).build()
), true, FetchVersion.ALL);
beforeRename.sort(Comparator.comparing(NamespaceFileMetadata::getVersion));
ArrayListTotal<NamespaceFileMetadata> afterRename = beforeRename
.map(nsFileMetadata -> {
String newPath;
if (nsFileMetadata.isDirectory()) {
newPath = normalizedTarget.toString().endsWith("/") ? normalizedTarget.toString() : normalizedTarget + "/";
} else {
newPath = normalizedTarget.toString();
}
return nsFileMetadata.toBuilder().path(newPath).build();
});
return afterRename.map(throwFunction(nsFileMetadata -> {
NamespaceFile beforeNamespaceFile = NamespaceFile.of(namespace, normalizedSource, nsFileMetadata.getVersion());
Path namespaceFilePath = beforeNamespaceFile.storagePath();
NamespaceFile afterNamespaceFile;
if (nsFileMetadata.isDirectory()) {
afterNamespaceFile = this.createDirectory(Path.of(nsFileMetadata.getPath()));
} else {
try (InputStream oldContent = storage.get(tenant, namespace, namespaceFilePath.toUri())) {
afterNamespaceFile = this.putFile(Path.of(nsFileMetadata.getPath()), oldContent, Conflicts.OVERWRITE).getFirst();
}
}
this.purge(NamespaceFile.of(namespace, normalizedSource, nsFileMetadata.getVersion()));
return Pair.of(beforeNamespaceFile, afterNamespaceFile);
}));
}
/**
* {@inheritDoc}
**/
@Override
public NamespaceFile get(final Path path) {
return NamespaceFile.of(namespace, path);
public NamespaceFile get(Path path) throws IOException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
int version = findByPath(normalizedPath).map(NamespaceFileMetadata::getVersion).orElse(1);
return NamespaceFile.of(namespace, normalizedPath, version);
}
public Path relativize(final URI uri) {
@@ -122,90 +217,225 @@ public class InternalNamespace implements Namespace {
* {@inheritDoc}
**/
@Override
public InputStream getFileContent(final Path path) throws IOException {
Path namespaceFilePath = NamespaceFile.of(namespace, path).storagePath();
public InputStream getFileContent(Path path, @Nullable Integer version) throws IOException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
// Throw if file not found OR if it's deleted
NamespaceFileMetadata namespaceFileMetadata = findByPath(normalizedPath, version).orElseThrow(() -> fileNotFound(normalizedPath, version));
Path namespaceFilePath = NamespaceFile.of(namespace, normalizedPath, namespaceFileMetadata.getVersion()).storagePath();
return storage.get(tenant, namespace, namespaceFilePath.toUri());
}
@Override
public FileAttributes getFileMetadata(Path path) throws IOException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
return findByPath(normalizedPath).map(NamespaceFileAttributes::new).orElseThrow(() -> fileNotFound(normalizedPath, null));
}
private FileNotFoundException fileNotFound(Path path, @Nullable Integer version) {
return new FileNotFoundException(Optional.ofNullable(version).map(v -> "Version " + v + " of file").orElse("File") + " '" + path + "' was not found in namespace '" + namespace + "'.");
}
private Optional<NamespaceFileMetadata> findByPath(Path path, boolean allowDeleted, @Nullable Integer version) throws IOException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
if (version != null) {
return namespaceFileMetadataRepository.find(Pageable.from(1, 1), tenant, List.of(
QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
QueryFilter.builder().field(QueryFilter.Field.PATH).operation(QueryFilter.Op.EQUALS).value(normalizedPath.toString()).build(),
QueryFilter.builder().field(QueryFilter.Field.VERSION).operation(QueryFilter.Op.EQUALS).value(version).build()
), allowDeleted, FetchVersion.ALL).stream().findFirst();
}
return namespaceFileMetadataRepository.findByPath(tenant, namespace, normalizedPath.toString())
.filter(namespaceFileMetadata -> allowDeleted || !namespaceFileMetadata.isDeleted());
}
private Optional<NamespaceFileMetadata> findByPath(Path path, boolean allowDeleted) throws IOException {
return findByPath(path, allowDeleted, null);
}
private Optional<NamespaceFileMetadata> findByPath(Path path, @Nullable Integer version) throws IOException {
return findByPath(path, false, version);
}
private Optional<NamespaceFileMetadata> findByPath(Path path) throws IOException {
return findByPath(path, null);
}
@Override
public boolean exists(Path path) throws IOException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
return findByPath(normalizedPath).isPresent();
}
/**
* {@inheritDoc}
**/
@Override
public NamespaceFile putFile(final Path path, final InputStream content, final Conflicts onAlreadyExist) throws IOException, URISyntaxException {
Path namespaceFilesPrefix = NamespaceFile.of(namespace, path).storagePath();
public List<NamespaceFile> putFile(final Path path, final InputStream content, final Conflicts onAlreadyExist) throws IOException, URISyntaxException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
Optional<NamespaceFileMetadata> inRepository = findByPath(normalizedPath, true);
int currentVersion = inRepository.map(NamespaceFileMetadata::getVersion).orElse(0);
NamespaceFile namespaceFile = NamespaceFile.of(namespace, normalizedPath, currentVersion + 1);
Path storagePath = namespaceFile.storagePath();
// Remove Windows letter
URI cleanUri = new URI(namespaceFilesPrefix.toUri().toString().replaceFirst("^file:///[a-zA-Z]:", ""));
final boolean exists = storage.exists(tenant, namespace, cleanUri);
URI cleanUri = new URI(storagePath.toUri().toString().replaceFirst("^file:///[a-zA-Z]:", ""));
return switch (onAlreadyExist) {
case OVERWRITE -> {
URI uri = storage.put(tenant, namespace, cleanUri, content);
NamespaceFile namespaceFile = new NamespaceFile(relativize(uri), uri, namespace);
if (exists) {
logger.debug(String.format(
"File '%s' overwritten into namespace '%s'.",
path,
namespace
));
} else {
logger.debug(String.format(
"File '%s' added to namespace '%s'.",
path,
namespace
));
}
yield namespaceFile;
List<NamespaceFile> createdFiles = new ArrayList<>();
if (inRepository.isEmpty()) {
storage.put(tenant, namespace, cleanUri, content);
createdFiles.addAll(mkDirs(normalizedPath.toString()));
namespaceFileMetadataRepository.save(
NamespaceFileMetadata.builder()
.tenantId(tenant)
.namespace(namespace)
.path(normalizedPath.toString())
.size(storage.getAttributes(tenant, namespace, cleanUri).getSize())
.build()
);
logger.debug(String.format(
"File '%s' added to namespace '%s'.",
normalizedPath,
namespace
));
createdFiles.add(namespaceFile);
} else if (onAlreadyExist == Conflicts.OVERWRITE || inRepository.get().isDeleted()) {
storage.put(tenant, namespace, cleanUri, content);
createdFiles.addAll(mkDirs(normalizedPath.toString()));
namespaceFileMetadataRepository.save(
inRepository.get().toBuilder().size(storage.getAttributes(tenant, namespace, cleanUri).getSize()).deleted(false).build()
);
if (inRepository.get().isDeleted()) {
logger.debug(String.format(
"File '%s' added to namespace '%s'.",
normalizedPath,
namespace
));
} else {
logger.debug(String.format(
"File '%s' overwritten into namespace '%s'.",
normalizedPath,
namespace
));
}
case ERROR -> {
if (!exists) {
URI uri = storage.put(tenant, namespace, namespaceFilesPrefix.toUri(), content);
yield new NamespaceFile(relativize(uri), uri, namespace);
} else {
throw new IOException(String.format(
"File '%s' already exists in namespace '%s' and conflict is set to %s",
path,
namespace,
Conflicts.ERROR
));
}
createdFiles.add(namespaceFile);
} else {
// At this point, the file exists and we have to decide what to do based on the conflict strategy
switch (onAlreadyExist) {
case ERROR -> throw new IOException(String.format(
"File '%s' already exists in namespace '%s' and conflict is set to %s",
normalizedPath,
namespace,
Conflicts.ERROR
));
case SKIP -> logger.debug(String.format(
"File '%s' already exists in namespace '%s' and conflict is set to %s. Skipping.",
normalizedPath,
namespace,
Conflicts.SKIP
));
}
case SKIP -> {
if (!exists) {
URI uri = storage.put(tenant, namespace, namespaceFilesPrefix.toUri(), content);
NamespaceFile namespaceFile = new NamespaceFile(relativize(uri), uri, namespace);
logger.debug(String.format(
"File '%s' added to namespace '%s'.",
path,
namespace
));
yield namespaceFile;
} else {
logger.debug(String.format(
"File '%s' already exists in namespace '%s' and conflict is set to %s. Skipping.",
path,
namespace,
Conflicts.SKIP
));
URI uri = URI.create(StorageContext.KESTRA_PROTOCOL + namespaceFilesPrefix);
yield new NamespaceFile(relativize(uri), uri, namespace);
}
}
};
}
return createdFiles;
}
/**
* Make all parent directories for a given path.
*/
private List<NamespaceFile> mkDirs(String path) throws IOException {
List<NamespaceFile> createdDirs = new ArrayList<>();
Optional<Path> maybeParentPath = Optional.empty();
while (
(maybeParentPath = Optional.ofNullable(NamespaceFileMetadata.parentPath(maybeParentPath.map(Path::toString).orElse(path))).map(Path::of)).isPresent()
&& !this.exists(maybeParentPath.get())
) {
this.createDirectory(maybeParentPath.get());
createdDirs.add(NamespaceFile.of(namespace, maybeParentPath.get().toString().endsWith("/") ? maybeParentPath.get().toString() : maybeParentPath.get() + "/", 1));
}
return createdDirs;
}
/**
* {@inheritDoc}
**/
@Override
public URI createDirectory(Path path) throws IOException {
return storage.createDirectory(tenant, namespace, NamespaceFile.of(namespace, path).storagePath().toUri());
public NamespaceFile createDirectory(Path path) throws IOException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
NamespaceFileMetadata nsFileMetadata = namespaceFileMetadataRepository.save(
NamespaceFileMetadata.builder()
.tenantId(tenant)
.namespace(namespace)
.path(normalizedPath.toString().endsWith("/") ? normalizedPath.toString() : normalizedPath + "/")
.size(0L)
.build()
);
storage.createDirectory(tenant, namespace, NamespaceFile.of(namespace, normalizedPath, 1).storagePath().toUri());
return NamespaceFile.fromMetadata(nsFileMetadata);
}
/**
* {@inheritDoc}
**/
@Override
public boolean delete(Path path) throws IOException {
return storage.delete(tenant, namespace, URI.create(path.toString().replace("\\","/")));
public List<NamespaceFile> delete(Path path) throws IOException {
final Path normalizedPath = NamespaceFile.normalize(path, true);
Optional<NamespaceFileMetadata> maybeNamespaceFileMetadata = namespaceFileMetadataRepository.find(Pageable.from(1, 1), tenant, List.of(
QueryFilter.builder().field(QueryFilter.Field.NAMESPACE).operation(QueryFilter.Op.EQUALS).value(namespace).build(),
QueryFilter.builder().field(QueryFilter.Field.PATH).operation(QueryFilter.Op.IN).value(List.of(normalizedPath.toString(), normalizedPath + "/")).build()
), false).stream().findFirst();
List<NamespaceFileMetadata> toDelete = Stream.concat(
this.children(normalizedPath.toString(), true).stream().map(NamespaceFileMetadata::toDeleted),
maybeNamespaceFileMetadata.map(NamespaceFileMetadata::toDeleted).stream()
).toList();
toDelete.forEach(namespaceFileMetadataRepository::save);
return toDelete.stream().map(NamespaceFile::fromMetadata).toList();
}
@Override
public boolean purge(NamespaceFile namespaceFile) throws IOException {
storage.delete(tenant, namespace, namespaceFile.storagePath().toUri());
namespaceFileMetadataRepository.purge(List.of(NamespaceFileMetadata.of(tenant, namespaceFile)));
return true;
}
/**
* {@inheritDoc}
*/
@Override
public Integer purge(List<NamespaceFile> namespaceFiles) throws IOException {
Integer purgedMetadataCount = this.namespaceFileMetadataRepository.purge(namespaceFiles.stream().map(namespaceFile -> NamespaceFileMetadata.of(tenant, namespaceFile)).toList());
long actualDeletedEntries = namespaceFiles.stream()
.map(NamespaceFile::storagePath)
.map(Path::toUri)
.map(throwFunction(uri -> this.storage.delete(tenant, namespace, uri)))
.filter(Boolean::booleanValue)
.count();
if (actualDeletedEntries != purgedMetadataCount) {
LOG.warn("Namespace Files Metadata purge reported {} deleted entries, but {} values were actually deleted from storage", purgedMetadataCount, actualDeletedEntries);
}
return purgedMetadataCount;
}
}

View File

@@ -1,15 +1,12 @@
package io.kestra.core.storages;
import io.kestra.core.services.FlowService;
import io.kestra.core.services.KVStoreService;
import io.kestra.core.storages.kv.InternalKVStore;
import io.kestra.core.storages.kv.KVStore;
import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
import io.kestra.core.services.NamespaceService;
import jakarta.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
@@ -33,7 +30,8 @@ public class InternalStorage implements Storage {
private final Logger logger;
private final StorageContext context;
private final StorageInterface storage;
private final FlowService flowService;
private final NamespaceFactory namespaceFactory;
private final NamespaceService namespaceService;
/**
* Creates a new {@link InternalStorage} instance.
@@ -41,8 +39,8 @@ public class InternalStorage implements Storage {
* @param context The storage context.
* @param storage The storage to delegate operations.
*/
public InternalStorage(StorageContext context, StorageInterface storage) {
this(LOG, context, storage, null);
public InternalStorage(StorageContext context, StorageInterface storage, NamespaceFactory namespaceFactory) {
this(LOG, context, storage, null, namespaceFactory);
}
/**
@@ -52,11 +50,12 @@ public class InternalStorage implements Storage {
* @param context The storage context.
* @param storage The storage to delegate operations.
*/
public InternalStorage(Logger logger, StorageContext context, StorageInterface storage, FlowService flowService) {
public InternalStorage(Logger logger, StorageContext context, StorageInterface storage, NamespaceService namespaceService, NamespaceFactory namespaceFactory) {
this.logger = logger;
this.context = context;
this.storage = storage;
this.flowService = flowService;
this.namespaceService = namespaceService;
this.namespaceFactory = namespaceFactory;
}
/**
@@ -64,7 +63,7 @@ public class InternalStorage implements Storage {
**/
@Override
public Namespace namespace() {
return new InternalNamespace(logger, context.getTenantId(), context.getNamespace(), storage);
return namespaceFactory.of(logger, context.getTenantId(), context.getNamespace(), storage);
}
/**
@@ -74,13 +73,13 @@ public class InternalStorage implements Storage {
public Namespace namespace(String namespace) {
boolean isExternalNamespace = !namespace.equals(context.getNamespace());
// Checks whether the contextual namespace is allowed to access the passed namespace.
if (isExternalNamespace && flowService != null) {
flowService.checkAllowedNamespace(
if (isExternalNamespace && namespaceService != null) {
namespaceService.checkAllowedNamespace(
context.getTenantId(), namespace, // requested Tenant/Namespace
context.getTenantId(), context.getNamespace() // from Tenant/Namespace
);
}
return new InternalNamespace(logger, context.getTenantId(), namespace, storage);
return namespaceFactory.of(logger, context.getTenantId(), namespace, storage);
}
/**
@@ -102,6 +101,13 @@ public class InternalStorage implements Storage {
}
@Override
public FileAttributes getAttributes(URI uri) throws IOException {
uriGuard(uri);
return this.storage.getAttributes(context.getTenantId(), context.getNamespace(), uri);
}
/**
* {@inheritDoc}
**/

View File

@@ -1,12 +1,22 @@
package io.kestra.core.storages;
import io.kestra.core.models.FetchVersion;
import io.kestra.core.models.QueryFilter;
import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
import io.kestra.core.repositories.ArrayListTotal;
import io.kestra.core.utils.PathMatcherPredicate;
import io.micronaut.data.model.Pageable;
import io.micronaut.data.model.Sort;
import jakarta.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Path;
import java.time.ZonedDateTime;
import java.util.Collections;
import java.util.List;
import java.util.function.Predicate;
@@ -16,6 +26,8 @@ import java.util.function.Predicate;
public interface Namespace {
String NAMESPACE_FILE_SCHEME = "nsfile";
ArrayListTotal<NamespaceFile> find(Pageable pageable, List<QueryFilter> filters, boolean allowDeleted, FetchVersion fetchVersion);
/**
* Gets the current namespace.
*
@@ -37,19 +49,25 @@ public interface Namespace {
*/
List<NamespaceFile> all() throws IOException;
/**
* Gets the URIs of all namespace files for the contextual namespace.
*
* @return The list of {@link URI}.
*/
List<NamespaceFile> all(boolean includeDirectories) throws IOException;
default List<NamespaceFile> all(String containing) throws IOException {
return this.all(containing, false);
}
/**
* Gets the URIs of all namespace files for the current namespace.
* Gets the URIs of all namespace files for the current namespace that contains the optional <code>containing</code> parameter.
*
* @return The list of {@link URI}.
*/
List<NamespaceFile> all(String prefix, boolean includeDirectories) throws IOException;
List<NamespaceFile> all(String containing, boolean includeDirectories) throws IOException;
/**
* Gets the URIs of all namespace files for the current namespace under the <code>parentPath</code>.
*
* @return The list of {@link URI}.
*/
List<NamespaceFileMetadata> children(String parentPath, boolean recursive) throws IOException;
List<Pair<NamespaceFile, NamespaceFile>> move(Path source, Path target) throws Exception;
/**
* Gets a {@link NamespaceFile} for the given path and the current namespace.
@@ -57,7 +75,7 @@ public interface Namespace {
* @param path the file path.
* @return a new {@link NamespaceFile}
*/
NamespaceFile get(Path path);
NamespaceFile get(Path path) throws IOException;
/**
* Retrieves the URIs of all namespace files for the current namespace matching the given predicate.
@@ -82,27 +100,45 @@ public interface Namespace {
return findAllFilesMatching(predicate);
}
/**
* Retrieves the content of the namespace file at the given path for the latest version.
*/
default InputStream getFileContent(Path path) throws IOException {
return getFileContent(path, null);
}
/**
* Retrieves the content of the namespace file at the given path.
*
* @param path the file path.
* @param version optionally a file version, otherwise will retrieve the latest.
* @return the {@link InputStream}.
* @throws IllegalArgumentException if the given {@link Path} is {@code null} or invalid.
* @throws IOException if an error happens while accessing the file.
*/
InputStream getFileContent(Path path) throws IOException;
InputStream getFileContent(Path path, @Nullable Integer version) throws IOException;
default NamespaceFile putFile(Path path, InputStream content) throws IOException, URISyntaxException {
/**
* Retrieves the metadata of the namespace file at the given path.
*
* @param path the file path.
* @return the {@link FileAttributes}.
*/
FileAttributes getFileMetadata(Path path) throws IOException;
boolean exists(Path path) throws IOException;
default List<NamespaceFile> putFile(Path path, InputStream content) throws IOException, URISyntaxException {
return putFile(path, content, Conflicts.OVERWRITE);
}
NamespaceFile putFile(Path path, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException;
List<NamespaceFile> putFile(Path path, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException;
default NamespaceFile putFile(NamespaceFile file, InputStream content) throws IOException, URISyntaxException {
default List<NamespaceFile> putFile(NamespaceFile file, InputStream content) throws IOException, URISyntaxException {
return putFile(file, content, Conflicts.OVERWRITE);
}
default NamespaceFile putFile(NamespaceFile file, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException {
default List<NamespaceFile> putFile(NamespaceFile file, InputStream content, Conflicts onAlreadyExist) throws IOException, URISyntaxException {
return putFile(Path.of(file.path()), content, onAlreadyExist);
}
@@ -110,39 +146,47 @@ public interface Namespace {
* Creates a new directory for the current namespace.
*
* @param path The {@link Path} of the directory.
* @return The URI of the directory in the Kestra's internal storage.
* @return The created namespace file.
* @throws IOException if an error happens while accessing the file.
*/
URI createDirectory(Path path) throws IOException;
NamespaceFile createDirectory(Path path) throws IOException;
/**
* Deletes any namespaces files at the given path.
* Deletes any namespaces file at the given path.
*
* @param file the {@link NamespaceFile} to be deleted.
* @throws IOException if an error happens while performing the delete operation.
*/
default boolean delete(NamespaceFile file) throws IOException {
default List<NamespaceFile> delete(NamespaceFile file) throws IOException {
return delete(Path.of(file.path()));
}
/**
* Deletes namespaces directories at the given path.
*
* @param file the {@link NamespaceFile} to be deleted.
* @throws IOException if an error happens while performing the delete operation.
*/
default boolean deleteDirectory(NamespaceFile file) throws IOException {
return delete(Path.of(file.path()));
}
/**
* Deletes any namespaces files at the given path.
* Soft-deletes any namespaces files at the given path.
*
* @param path the path to be deleted.
* @return {@code true} if the file was deleted by this method; {@code false} if the file could not be deleted because it did not exist
* @return the list of namespace files that got deleted. There can be multiple files if a directory is deleted as its whole content will be.
* @throws IOException if an error happens while performing the delete operation.
*/
boolean delete(Path path) throws IOException;
List<NamespaceFile> delete(Path path) throws IOException;
/**
* Hard-deletes any namespaces files.
*
* @param namespaceFile the namespace file to be purged.
* @return {@code true} if the file was purged by this method; {@code false} if the file could not be deleted because it did not exist
* @throws IOException if an error happens while performing the delete operation.
*/
boolean purge(NamespaceFile namespaceFile) throws IOException;
/**
* Hard-deletes all provided namespaces files.
*
* @param namespaceFiles the namespace files to be purged.
* @return the amount of files that were purged.
* @throws IOException if an error happens while performing the delete operation.
*/
Integer purge(List<NamespaceFile> namespaceFiles) throws IOException;
/**
* Checks if a directory is empty.

View File

@@ -0,0 +1,20 @@
package io.kestra.core.storages;
import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
import org.slf4j.Logger;
@Singleton
public class NamespaceFactory {
@Inject
private NamespaceFileMetadataRepositoryInterface namespaceFileMetadataRepositoryInterface;
public Namespace of(String tenantId, String namespace, StorageInterface storageInterface) {
return new InternalNamespace(tenantId, namespace, storageInterface, namespaceFileMetadataRepositoryInterface);
}
public Namespace of(Logger logger, String tenantId, String namespace, StorageInterface storageInterface) {
return new InternalNamespace(logger, tenantId, namespace, storageInterface, namespaceFileMetadataRepositoryInterface);
}
}

View File

@@ -1,11 +1,14 @@
package io.kestra.core.storages;
import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
import io.kestra.core.utils.WindowsUtils;
import jakarta.annotation.Nullable;
import java.net.URI;
import java.nio.file.Path;
import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
* Represents a NamespaceFile object.
@@ -13,15 +16,22 @@ import java.util.Objects;
* @param path The path of file relative to the namespace.
* @param uri The URI of the namespace file in the Kestra's internal storage.
* @param namespace The namespace of the file.
* @param version The version of the file.
*/
public record NamespaceFile(
String path,
URI uri,
String namespace
String namespace,
int version
) {
private static final Pattern capturePathWithoutVersion = Pattern.compile("(.*)(?:\\.v\\d+)?$");
public NamespaceFile(Path path, URI uri, String namespace) {
this(path.toString(), uri, namespace);
this(path.toString(), uri, namespace, 1);
}
public NamespaceFile(String path, URI uri, String namespace) {
this(path, uri, namespace, 1);
}
/**
@@ -33,7 +43,19 @@ public record NamespaceFile(
* @return a new {@link NamespaceFile} object
*/
public static NamespaceFile of(final String namespace) {
return of(namespace, (Path) null);
return of(namespace, (Path) null, 1);
}
public static NamespaceFile of(final String namespace, final URI uri) {
return of(namespace, uri, 1);
}
public static NamespaceFile fromMetadata(final NamespaceFileMetadata metadata) {
return of(
metadata.getNamespace(),
Path.of(metadata.getPath()),
metadata.getVersion()
);
}
/**
@@ -43,9 +65,9 @@ public record NamespaceFile(
* @param namespace The namespace - cannot be {@code null}.
* @return a new {@link NamespaceFile} object
*/
public static NamespaceFile of(final String namespace, @Nullable final URI uri) {
public static NamespaceFile of(final String namespace, @Nullable final URI uri, int version) {
if (uri == null || uri.equals(URI.create("/"))) {
return of(namespace, (Path) null);
return of(namespace, (Path) null, version);
}
Path path = Path.of(WindowsUtils.windowsToUnixPath(uri.getPath()));
@@ -61,9 +83,9 @@ public record NamespaceFile(
"Invalid Kestra URI. Expected prefix for namespace '%s', but was %s.", namespace, uri)
);
}
namespaceFile = of(namespace, Path.of(StorageContext.namespaceFilePrefix(namespace)).relativize(path));
namespaceFile = of(namespace, Path.of(StorageContext.namespaceFilePrefix(namespace)).relativize(path), version);
} else {
namespaceFile = of(namespace, path);
namespaceFile = of(namespace, path, version);
}
boolean trailingSlash = uri.toString().endsWith("/");
@@ -75,10 +97,15 @@ public record NamespaceFile(
return new NamespaceFile(
namespaceFile.path,
URI.create(namespaceFile.uri.toString() + "/"),
namespaceFile.namespace
namespaceFile.namespace,
version
);
}
public static NamespaceFile of(final String namespace, final Path path) {
return of(namespace, path, 1);
}
/**
* Static factory method for constructing a new {@link NamespaceFile} object.
*
@@ -86,31 +113,61 @@ public record NamespaceFile(
* @param namespace The namespace - cannot be {@code null}.
* @return a new {@link NamespaceFile} object
*/
public static NamespaceFile of(final String namespace, @Nullable final Path path) {
public static NamespaceFile of(final String namespace, @Nullable final Path path, int version) {
Objects.requireNonNull(namespace, "namespace cannot be null");
if (path == null || path.equals(Path.of("/"))) {
return new NamespaceFile(
"",
URI.create(StorageContext.KESTRA_PROTOCOL + StorageContext.namespaceFilePrefix(namespace) + "/"),
namespace
namespace,
// Directory always has a single version
1
);
}
return of(namespace, path.toString(), version);
}
public static NamespaceFile of(String namespace, String path, int version) {
Path namespacePrefixPath = Path.of(StorageContext.namespaceFilePrefix(namespace));
Path filePath = path.normalize();
if (filePath.isAbsolute()) {
filePath = filePath.getRoot().relativize(filePath);
}
// Need to remove starting trailing slash for Windows
String pathWithoutTrailingSlash = path.toString().replaceFirst("^[.]*[\\\\|/]+", "");
String pathWithoutLeadingSlash = path.replaceFirst("^[.]*[\\\\|/]+", "");
version = NamespaceFile.isDirectory(pathWithoutLeadingSlash) ? 1 : version;
String storagePath = pathWithoutLeadingSlash;
if (!pathWithoutLeadingSlash.endsWith("/") && version > 1) {
storagePath += ".v" + version;
}
return new NamespaceFile(
pathWithoutTrailingSlash,
URI.create(StorageContext.KESTRA_PROTOCOL + namespacePrefixPath.resolve(pathWithoutTrailingSlash).toString().replace("\\","/")),
namespace
pathWithoutLeadingSlash,
URI.create(StorageContext.KESTRA_PROTOCOL + namespacePrefixPath.resolve(storagePath).toString().replace("\\", "/")),
namespace,
version
);
}
public static Path normalize(String pathStr, boolean withLeadingSlash) {
return normalize(Path.of(pathStr), withLeadingSlash);
}
public static Path normalize(Path path, boolean withLeadingSlash) {
if (path == null) {
return Path.of("/");
}
if (withLeadingSlash && !path.toString().startsWith("/")) {
return Path.of("/" + path);
}
if (!withLeadingSlash && path.toString().startsWith("/")) {
return Path.of(path.toString().substring(1));
}
return path;
}
/**
* Returns the path of file relative to the namespace.
*
@@ -118,17 +175,13 @@ public record NamespaceFile(
* @return The path.
*/
public Path path(boolean withLeadingSlash) {
final String strPath = path.toString();
if (!withLeadingSlash) {
if (strPath.startsWith("/")) {
return Path.of(strPath.substring(1));
}
} else {
if (!strPath.startsWith("/")) {
return Path.of("/").resolve(path);
}
String strPath = path;
Matcher matcher = capturePathWithoutVersion.matcher(strPath);
if (matcher.matches()) {
strPath = matcher.group(1);
}
return Path.of(path);
return normalize(Path.of(strPath), withLeadingSlash);
}
/**
@@ -147,8 +200,12 @@ public record NamespaceFile(
*
* @return {@code true} if this namespace file is a directory.
*/
public static boolean isDirectory(String path) {
return path.endsWith("/");
}
public boolean isDirectory() {
return uri.toString().endsWith("/");
return isDirectory(uri.toString());
}
/**

View File

@@ -0,0 +1,54 @@
package io.kestra.core.storages;
import io.kestra.core.models.namespaces.files.NamespaceFileMetadata;
import java.io.File;
import java.io.IOException;
import java.time.Instant;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
public class NamespaceFileAttributes implements FileAttributes {
private final NamespaceFileMetadata namespaceFileMetadata;
public NamespaceFileAttributes(NamespaceFileMetadata namespaceFileMetadata) {
this.namespaceFileMetadata = namespaceFileMetadata;
}
@Override
public String getFileName() {
String name = new File(namespaceFileMetadata.getPath()).getName();
if (name.isEmpty()) {
return "_files";
}
return name;
}
@Override
public long getLastModifiedTime() {
return Optional.ofNullable(namespaceFileMetadata.getUpdated()).map(Instant::toEpochMilli).orElse(0L);
}
@Override
public long getCreationTime() {
return Optional.ofNullable(namespaceFileMetadata.getCreated()).map(Instant::toEpochMilli).orElse(0L);
}
@Override
public FileType getType() {
return namespaceFileMetadata.getPath().endsWith("/") ? FileType.Directory : FileType.File;
}
@Override
public long getSize() {
return namespaceFileMetadata.getSize();
}
@Override
public Map<String, String> getMetadata() throws IOException {
return Collections.emptyMap();
}
}

View File

@@ -0,0 +1,3 @@
package io.kestra.core.storages;
public record NamespaceFileRevision(Integer revision) {}

View File

@@ -1,8 +1,10 @@
package io.kestra.core.storages;
import io.kestra.core.annotations.Retryable;
import jakarta.annotation.Nullable;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
@@ -46,6 +48,15 @@ public interface Storage {
*/
InputStream getFile(URI uri) throws IOException;
/**
* Retrieves the metadata attributes for the given URI.
*
* @param uri the URI of the object
* @return the file attributes
* @throws IOException if the attributes cannot be retrieved
*/
FileAttributes getAttributes(URI uri) throws IOException;
/**
* Deletes the file for the given URI.
* @param uri the file URI.

View File

@@ -13,6 +13,7 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.nio.file.NoSuchFileException;
import java.util.List;
/**
@@ -52,7 +53,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return an InputStream to read the object's contents
* @throws IOException if the object cannot be read
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
InputStream get(String tenantId, @Nullable String namespace, URI uri) throws IOException;
/**
@@ -64,7 +65,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return an InputStream to read the object's contents
* @throws IOException if the object cannot be read
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
InputStream getInstanceResource(@Nullable String namespace, URI uri) throws IOException;
/**
@@ -76,7 +77,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return the storage object with metadata
* @throws IOException if the object cannot be retrieved
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
StorageObject getWithMetadata(String tenantId, @Nullable String namespace, URI uri) throws IOException;
/**
@@ -89,7 +90,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return a list of matching object URIs
* @throws IOException if the listing fails
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
List<URI> allByPrefix(String tenantId, @Nullable String namespace, URI prefix, boolean includeDirectories) throws IOException;
/**
@@ -101,7 +102,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return a list of file attributes
* @throws IOException if the listing fails
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
List<FileAttributes> list(String tenantId, @Nullable String namespace, URI uri) throws IOException;
/**
@@ -113,7 +114,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return a list of file attributes
* @throws IOException if the listing fails
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
List<FileAttributes> listInstanceResource(@Nullable String namespace, URI uri) throws IOException;
/**
@@ -159,7 +160,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return the file attributes
* @throws IOException if the attributes cannot be retrieved
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
FileAttributes getAttributes(String tenantId, @Nullable String namespace, URI uri) throws IOException;
/**
@@ -171,7 +172,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return the file attributes
* @throws IOException if the attributes cannot be retrieved
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
FileAttributes getInstanceAttributes(@Nullable String namespace, URI uri) throws IOException;
/**
@@ -288,7 +289,7 @@ public interface StorageInterface extends AutoCloseable, Plugin {
* @return the URI of the moved object
* @throws IOException if moving fails
*/
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class})
@Retryable(includes = {IOException.class}, excludes = {FileNotFoundException.class, NoSuchFileException.class})
URI move(String tenantId, @Nullable String namespace, URI from, URI to) throws IOException;
/**

View File

@@ -6,6 +6,10 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;
import java.util.function.Supplier;
/**
* @deprecated use {@link org.awaitility.Awaitility} instead
*/
@Deprecated
public class Await {
private static final Duration defaultSleep = Duration.ofMillis(100);

View File

@@ -14,8 +14,6 @@ import lombok.extern.slf4j.Slf4j;
@Singleton
@Slf4j
public class ExecutorsUtils {
@Inject
private ThreadMainFactoryBuilder threadFactoryBuilder;
@Inject
private MeterRegistry meterRegistry;
@@ -24,7 +22,7 @@ public class ExecutorsUtils {
return this.wrap(
name,
Executors.newCachedThreadPool(
threadFactoryBuilder.build(name + "_%d")
ThreadMainFactoryBuilder.build(name + "_%d")
)
);
}
@@ -36,7 +34,7 @@ public class ExecutorsUtils {
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
threadFactoryBuilder.build(name + "_%d")
ThreadMainFactoryBuilder.build(name + "_%d")
);
threadPoolExecutor.allowCoreThreadTimeOut(true);
@@ -51,7 +49,7 @@ public class ExecutorsUtils {
return this.wrap(
name,
Executors.newSingleThreadExecutor(
threadFactoryBuilder.build(name + "_%d")
ThreadMainFactoryBuilder.build(name + "_%d")
)
);
}
@@ -60,7 +58,7 @@ public class ExecutorsUtils {
return this.wrap(
name,
Executors.newSingleThreadScheduledExecutor(
threadFactoryBuilder.build(name + "_%d")
ThreadMainFactoryBuilder.build(name + "_%d")
)
);
}

View File

@@ -65,10 +65,9 @@ public class ListUtils {
}
public static List<String> convertToListString(Object object){
if (object instanceof List<?> list && (list.isEmpty() || list.getFirst() instanceof String)) {
return (List<String>) list;
} else {
throw new IllegalArgumentException("%s in not an instance of List of String".formatted(object));
}
return convertToList(object)
.stream()
.map(Object::toString)
.toList();
}
}

View File

@@ -1,38 +1,27 @@
package io.kestra.core.services;
package io.kestra.core.utils;
import io.kestra.core.models.executions.Execution;
import io.kestra.core.models.executions.LogEntry;
import io.kestra.core.models.executions.TaskRun;
import io.kestra.core.models.flows.FlowId;
import io.kestra.core.models.triggers.TriggerContext;
import io.kestra.core.repositories.LogRepositoryInterface;
import io.micronaut.data.model.Pageable;
import io.micronaut.data.model.Sort;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;
import java.time.ZonedDateTime;
import java.util.List;
/**
* Utility class for logging
*/
public final class Logs {
@Singleton
public class LogService {
private static final String FLOW_PREFIX_WITH_TENANT = "[tenant: {}] [namespace: {}] [flow: {}] ";
private static final String EXECUTION_PREFIX_WITH_TENANT = FLOW_PREFIX_WITH_TENANT + "[execution: {}] ";
private static final String TRIGGER_PREFIX_WITH_TENANT = FLOW_PREFIX_WITH_TENANT + "[trigger: {}] ";
private static final String TASKRUN_PREFIX_WITH_TENANT = FLOW_PREFIX_WITH_TENANT + "[task: {}] [execution: {}] [taskrun: {}] ";
private Logs() {}
private final LogRepositoryInterface logRepository;
@Inject
public LogService(LogRepositoryInterface logRepository) {
this.logRepository = logRepository;
}
public void logExecution(FlowId flow, Logger logger, Level level, String message, Object... args) {
public static void logExecution(FlowId flow, Logger logger, Level level, String message, Object... args) {
String finalMsg = FLOW_PREFIX_WITH_TENANT + message;
Object[] executionArgs = new Object[] { flow.getTenantId(), flow.getNamespace(), flow.getId() };
Object[] finalArgs = ArrayUtils.addAll(executionArgs, args);
@@ -40,37 +29,37 @@ public class LogService {
}
/**
* Log an execution via the execution logger named: 'execution.{flowId}'.
* Log an {@link Execution} via the execution logger named: 'execution.{flowId}'.
*/
public void logExecution(Execution execution, Level level, String message, Object... args) {
public static void logExecution(Execution execution, Level level, String message, Object... args) {
Logger logger = logger(execution);
logExecution(execution, logger, level, message, args);
}
public void logExecution(Execution execution, Logger logger, Level level, String message, Object... args) {
public static void logExecution(Execution execution, Logger logger, Level level, String message, Object... args) {
Object[] executionArgs = new Object[] { execution.getTenantId(), execution.getNamespace(), execution.getFlowId(), execution.getId() };
Object[] finalArgs = ArrayUtils.addAll(executionArgs, args);
logger.atLevel(level).log(EXECUTION_PREFIX_WITH_TENANT + message, finalArgs);
}
/**
* Log a trigger via the trigger logger named: 'trigger.{flowId}.{triggereId}'.
* Log a {@link TriggerContext} via the trigger logger named: 'trigger.{flowId}.{triggereId}'.
*/
public void logTrigger(TriggerContext triggerContext, Level level, String message, Object... args) {
public static void logTrigger(TriggerContext triggerContext, Level level, String message, Object... args) {
Logger logger = logger(triggerContext);
logTrigger(triggerContext, logger, level, message, args);
}
public void logTrigger(TriggerContext triggerContext, Logger logger, Level level, String message, Object... args) {
public static void logTrigger(TriggerContext triggerContext, Logger logger, Level level, String message, Object... args) {
Object[] executionArgs = new Object[] { triggerContext.getTenantId(), triggerContext.getNamespace(), triggerContext.getFlowId(), triggerContext.getTriggerId() };
Object[] finalArgs = ArrayUtils.addAll(executionArgs, args);
logger.atLevel(level).log(TRIGGER_PREFIX_WITH_TENANT + message, finalArgs);
}
/**
* Log a taskRun via the taskRun logger named: 'task.{flowId}.{taskId}'.
* Log a {@link TaskRun} via the taskRun logger named: 'task.{flowId}.{taskId}'.
*/
public void logTaskRun(TaskRun taskRun, Level level, String message, Object... args) {
public static void logTaskRun(TaskRun taskRun, Level level, String message, Object... args) {
String prefix = TASKRUN_PREFIX_WITH_TENANT;
String finalMsg = taskRun.getValue() == null ? prefix + message : prefix + "[value: {}] " + message;
Object[] executionArgs = new Object[] { taskRun.getTenantId(), taskRun.getNamespace(), taskRun.getFlowId(), taskRun.getTaskId(), taskRun.getExecutionId(), taskRun.getId() };
@@ -82,31 +71,19 @@ public class LogService {
logger.atLevel(level).log(finalMsg, finalArgs);
}
public int purge(String tenantId, String namespace, String flowId, String executionId, List<Level> logLevels, ZonedDateTime startDate, ZonedDateTime endDate) {
return logRepository.deleteByQuery(tenantId, namespace, flowId, executionId, logLevels, startDate, endDate);
}
/**
* Fetch the error logs of an execution.
* Will limit the results to the first 25 error logs, ordered by timestamp asc.
*/
public List<LogEntry> errorLogs(String tenantId, String executionId) {
return logRepository.findByExecutionId(tenantId, executionId, Level.ERROR, Pageable.from(1, 25, Sort.of(Sort.Order.asc("timestamp"))));
}
private Logger logger(TaskRun taskRun) {
private static Logger logger(TaskRun taskRun) {
return LoggerFactory.getLogger(
"task." + taskRun.getFlowId() + "." + taskRun.getTaskId()
);
}
private Logger logger(TriggerContext triggerContext) {
private static Logger logger(TriggerContext triggerContext) {
return LoggerFactory.getLogger(
"trigger." + triggerContext.getFlowId() + "." + triggerContext.getTriggerId()
);
}
private Logger logger(Execution execution) {
private static Logger logger(Execution execution) {
return LoggerFactory.getLogger(
"execution." + execution.getFlowId()
);

View File

@@ -32,10 +32,12 @@ public class NamespaceFilesUtils {
private ExecutorsUtils executorsUtils;
private ExecutorService executorService;
private int maxThreads;
@PostConstruct
public void postConstruct() {
this.executorService = executorsUtils.maxCachedThreadPool(Math.max(Runtime.getRuntime().availableProcessors() * 4, 32), "namespace-file");
this.maxThreads = Math.max(Runtime.getRuntime().availableProcessors() * 4, 32);
this.executorService = executorsUtils.maxCachedThreadPool(maxThreads, "namespace-file");
}
public void loadNamespaceFiles(
@@ -63,7 +65,11 @@ public class NamespaceFilesUtils {
matchedNamespaceFiles.addAll(files);
}
// Use half of the available threads to avoid impacting concurrent tasks
int parallelism = maxThreads / 2;
Flux.fromIterable(matchedNamespaceFiles)
.parallel(parallelism)
.runOn(Schedulers.fromExecutorService(executorService))
.doOnNext(throwConsumer(nsFile -> {
InputStream content = runContext.storage().getFile(nsFile.uri());
Path path = folderPerNamespace ?
@@ -71,7 +77,7 @@ public class NamespaceFilesUtils {
Path.of(nsFile.path());
runContext.workingDir().putFile(path, content, fileExistComportment);
}))
.publishOn(Schedulers.fromExecutorService(executorService))
.sequential()
.blockLast();
Duration duration = stopWatch.getDuration();

View File

@@ -17,36 +17,36 @@ import org.slf4j.Logger;
import java.io.Serial;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.function.BiPredicate;
import java.util.function.Function;
import java.util.function.Predicate;
import jakarta.inject.Singleton;
public final class RetryUtils {
private RetryUtils() {
// utility class pattern
}
@Singleton
public class RetryUtils {
public <T, E extends Throwable> Instance<T, E> of() {
public static <T, E extends Throwable> Instance<T, E> of() {
return Instance.<T, E>builder()
.build();
}
public <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy) {
public static <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy) {
return Instance.<T, E>builder()
.policy(policy)
.build();
}
public <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Function<RetryFailed, E> failureFunction) {
public static <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Function<RetryFailed, E> failureFunction) {
return Instance.<T, E>builder()
.policy(policy)
.failureFunction(failureFunction)
.build();
}
public <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Logger logger) {
public static <T, E extends Throwable> Instance<T, E> of(AbstractRetry policy, Logger logger) {
return Instance.<T, E>builder()
.policy(policy)
.logger(logger)
@@ -199,7 +199,6 @@ public class RetryUtils {
private final int attemptCount;
private final Duration elapsedTime;
private final Instant startTime;
public <T> RetryFailed(ExecutionAttemptedEvent<? extends T> event) {
super(
@@ -210,7 +209,6 @@ public class RetryUtils {
this.attemptCount = event.getAttemptCount();
this.elapsedTime = event.getElapsedTime();
this.startTime = event.getStartTime().get();
}
}
}

View File

@@ -3,18 +3,18 @@ package io.kestra.core.utils;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.ThreadFactory;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
@Singleton
public class ThreadMainFactoryBuilder {
@Inject
private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
public ThreadFactory build(String name) {
public final class ThreadMainFactoryBuilder {
private ThreadMainFactoryBuilder() {
// utility class pattern
}
public static ThreadFactory build(String name) {
return new ThreadFactoryBuilder()
.setNameFormat(name)
.setUncaughtExceptionHandler(this.uncaughtExceptionHandler)
.setUncaughtExceptionHandler(ThreadUncaughtExceptionHandler.INSTANCE)
.build();
}
}

View File

@@ -1,27 +1,21 @@
package io.kestra.core.utils;
import io.micronaut.context.ApplicationContext;
import io.kestra.core.contexts.KestraContext;
import lombok.extern.slf4j.Slf4j;
import java.lang.Thread.UncaughtExceptionHandler;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
@Slf4j
@Singleton
public final class ThreadUncaughtExceptionHandlers implements UncaughtExceptionHandler {
@Inject
private ApplicationContext applicationContext;
private final Runtime runtime = Runtime.getRuntime();
public final class ThreadUncaughtExceptionHandler implements UncaughtExceptionHandler {
public static final UncaughtExceptionHandler INSTANCE = new ThreadUncaughtExceptionHandler();
@Override
public void uncaughtException(Thread t, Throwable e) {
boolean isTest = applicationContext.getEnvironment().getActiveNames().contains("test");
boolean isTest = KestraContext.getContext().getEnvironments().contains("test");
try {
// cannot use FormattingLogger due to a dependency loop
log.error("Caught an exception in {}. " + (isTest ? "Keeping it running for test." : "Shutting down."), t, e);
log.error("Caught an exception in {}. {}", t, isTest ? "Keeping it running for test." : "Shutting down.", e);
} catch (Throwable errorInLogging) {
// If logging fails, e.g. due to missing memory, at least try to log the
// message and the cause for the failed logging.
@@ -29,8 +23,8 @@ public final class ThreadUncaughtExceptionHandlers implements UncaughtExceptionH
System.err.println(errorInLogging.getMessage());
} finally {
if (!isTest) {
applicationContext.close();
runtime.exit(1);
KestraContext.getContext().shutdown();
Runtime.getRuntime().exit(1);
}
}
}

View File

@@ -0,0 +1,16 @@
package io.kestra.core.validations;
import io.kestra.core.validations.validator.FilesVersionBehaviorValidator;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import jakarta.validation.Constraint;
import jakarta.validation.Payload;
@Retention(RetentionPolicy.RUNTIME)
@Constraint(validatedBy = FilesVersionBehaviorValidator.class)
public @interface FilesVersionBehaviorValidation {
String message() default "invalid `version` behavior configuration";
Class<?>[] groups() default {};
Class<? extends Payload>[] payload() default {};
}

View File

@@ -0,0 +1,35 @@
package io.kestra.core.validations.validator;
import io.kestra.core.validations.FilesVersionBehaviorValidation;
import io.kestra.core.validations.KvVersionBehaviorValidation;
import io.kestra.plugin.core.namespace.Version;
import io.micronaut.core.annotation.AnnotationValue;
import io.micronaut.core.annotation.Introspected;
import io.micronaut.core.annotation.NonNull;
import io.micronaut.core.annotation.Nullable;
import io.micronaut.validation.validator.constraints.ConstraintValidator;
import io.micronaut.validation.validator.constraints.ConstraintValidatorContext;
import jakarta.inject.Singleton;
@Singleton
@Introspected
public class FilesVersionBehaviorValidator implements ConstraintValidator<FilesVersionBehaviorValidation, Version> {
@Override
public boolean isValid(
@Nullable Version value,
@NonNull AnnotationValue<FilesVersionBehaviorValidation> annotationMetadata,
@NonNull ConstraintValidatorContext context) {
if (value == null) {
return true;
}
if (value.getBefore() != null && value.getKeepAmount() != null) {
context.disableDefaultConstraintViolation();
context.buildConstraintViolationWithTemplate("Cannot set both 'before' and 'keepAmount' properties")
.addConstraintViolation();
return false;
}
return true;
}
}

View File

@@ -6,6 +6,7 @@ import io.kestra.core.models.flows.Input;
import io.kestra.core.models.tasks.ExecutableTask;
import io.kestra.core.models.tasks.Task;
import io.kestra.core.services.FlowService;
import io.kestra.core.services.NamespaceService;
import io.kestra.core.utils.ListUtils;
import io.kestra.core.validations.FlowValidation;
import io.micronaut.core.annotation.AnnotationValue;
@@ -52,6 +53,9 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
@Inject
private FlowService flowService;
@Inject
private NamespaceService namespaceService;
@Override
public boolean isValid(
@Nullable Flow value,
@@ -67,7 +71,7 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
violations.add("Flow id is a reserved keyword: " + value.getId() + ". List of reserved keywords: " + String.join(", ", RESERVED_FLOW_IDS));
}
if (flowService.requireExistingNamespace(value.getTenantId(), value.getNamespace())) {
if (namespaceService.requireExistingNamespace(value.getTenantId(), value.getNamespace())) {
violations.add("Namespace '" + value.getNamespace() + "' does not exist but is required to exist before a flow can be created in it.");
}

View File

@@ -79,20 +79,30 @@ public class TimeBetween extends Condition implements ScheduleCondition {
RunContext runContext = conditionContext.getRunContext();
Map<String, Object> variables = conditionContext.getVariables();
String dateRendered = runContext.render(date).as(String.class, variables).orElseThrow();
// cache must be skipped for date rendering as the value can change for each test
String dateRendered = runContext.render(date).skipCache().as(String.class, variables).orElseThrow();
OffsetTime currentDate = DateUtils.parseZonedDateTime(dateRendered).toOffsetDateTime().toOffsetTime();
OffsetTime beforeRendered = runContext.render(before).as(OffsetTime.class, variables).orElse(null);
OffsetTime afterRendered = runContext.render(after).as(OffsetTime.class, variables).orElse(null);
if (beforeRendered != null && afterRendered != null) {
return currentDate.isAfter(afterRendered) && currentDate.isBefore(beforeRendered);
// Case 1: Normal range (e.g., 16:00 -> 20:00)
if (afterRendered.isBefore(beforeRendered)) {
return currentDate.isAfter(afterRendered) && currentDate.isBefore(beforeRendered);
// Case 2: Cross-midnight range (e.g., 22:00 -> 02:00)
} else {
return currentDate.isAfter(afterRendered) || currentDate.isBefore(beforeRendered);
}
} else if (beforeRendered != null) {
return currentDate.isBefore(beforeRendered);
} else if (afterRendered != null) {
return currentDate.isAfter(afterRendered);
} else {
throw new IllegalConditionEvaluation("Invalid condition with no before nor after");
throw new IllegalConditionEvaluation("Invalid condition: no 'before' or 'after' value defined");
}
}
}

View File

@@ -1,7 +1,6 @@
package io.kestra.plugin.core.dashboard.chart;
import com.fasterxml.jackson.annotation.JsonInclude;
import io.kestra.core.models.annotations.Example;
import io.kestra.core.models.annotations.Plugin;
import io.kestra.core.models.dashboards.ColumnDescriptor;
@@ -21,34 +20,33 @@ import lombok.experimental.SuperBuilder;
@EqualsAndHashCode
@Schema(
title = "Show proportions and distributions using pie charts."
)
)
@Plugin(
examples = {
@Example(
title = "Display a pie chart with Executions per State.",
full = true,
code = { """
code = """
charts:
- id: executions_pie
type: io.kestra.plugin.core.dashboard.chart.Pie
chartOptions:
displayName: Total Executions
description: Total executions per state
legend:
enabled: true
colorByColumn: state
data:
type: io.kestra.plugin.core.dashboard.data.Executions
columns:
state:
field: STATE
total:
agg: COUNT
- id: executions_pie
type: io.kestra.plugin.core.dashboard.chart.Pie
chartOptions:
displayName: Total Executions
description: Total executions per state
legend:
enabled: true
colorByColumn: state
data:
type: io.kestra.plugin.core.dashboard.data.Executions
columns:
state:
field: STATE
total:
agg: COUNT
"""
}
)
}
)
)
public class Pie<F extends Enum<F>, D extends DataFilter<F, ? extends ColumnDescriptor<F>>> extends DataChart<PieOption, D> {
@Override
public Integer minNumberOfAggregations() {

View File

@@ -1,7 +1,6 @@
package io.kestra.plugin.core.dashboard.chart;
import com.fasterxml.jackson.annotation.JsonInclude;
import io.kestra.core.models.annotations.Example;
import io.kestra.core.models.annotations.Plugin;
import io.kestra.core.models.dashboards.DataFilter;
@@ -21,33 +20,32 @@ import lombok.experimental.SuperBuilder;
@EqualsAndHashCode
@Schema(
title = "Display structured data in a clear, sortable table."
)
)
@Plugin(
examples = {
@Example(
title = "Display a table with Log counts for each level by Namespace.",
full = true,
code = { """
code = """
charts:
- id: table_logs
- id: table_logs
type: io.kestra.plugin.core.dashboard.chart.Table
chartOptions:
displayName: Log count by level for filtered namespace
displayName: Log count by level for filtered namespace
data:
type: io.kestra.plugin.core.dashboard.data.Logs
columns:
level:
field: LEVEL
count:
agg: COUNT
where:
- field: NAMESPACE
type: IN
values:
- dev_graph
- prod_graph
type: io.kestra.plugin.core.dashboard.data.Logs
columns:
level:
field: LEVEL
count:
agg: COUNT
where:
- field: NAMESPACE
type: IN
values:
- dev_graph
- prod_graph
"""
}
)
}
)

View File

@@ -1,7 +1,6 @@
package io.kestra.plugin.core.dashboard.chart;
import com.fasterxml.jackson.annotation.JsonInclude;
import io.kestra.core.models.annotations.Example;
import io.kestra.core.models.annotations.Plugin;
import io.kestra.core.models.dashboards.DataFilter;
@@ -23,42 +22,41 @@ import lombok.experimental.SuperBuilder;
@TimeSeriesChartValidation
@Schema(
title = "Track trends over time with dynamic time series charts."
)
)
@Plugin(
examples = {
@Example(
title = "Display a chart with Executions over the last week.",
full = true,
code = { """
code = """
charts:
- id: executions_timeseries
type: io.kestra.plugin.core.dashboard.chart.TimeSeries
chartOptions:
displayName: Total Executions
description: Executions last week
legend:
enabled: true
column: date
colorByColumn: state
displayName: Total Executions
description: Executions last week
legend:
enabled: true
column: date
colorByColumn: state
data:
type: io.kestra.plugin.core.dashboard.data.Executions
columns:
date:
field: START_DATE
displayName: Date
state:
field: STATE
total:
displayName: Executions
agg: COUNT
graphStyle: BARS
duration:
displayName: Duration
field: DURATION
agg: SUM
graphStyle: LINES
type: io.kestra.plugin.core.dashboard.data.Executions
columns:
date:
field: START_DATE
displayName: Date
state:
field: STATE
total:
displayName: Executions
agg: COUNT
graphStyle: BARS
duration:
displayName: Duration
field: DURATION
agg: SUM
graphStyle: LINES
"""
}
)
}
)

View File

@@ -30,28 +30,27 @@ import lombok.experimental.SuperBuilder;
@Example(
title = "Display a chart with a Executions per Namespace broken out by State.",
full = true,
code = { """
charts:
- id: executions_per_namespace_bars
type: io.kestra.plugin.core.dashboard.chart.Bar
chartOptions:
displayName: Executions (per namespace)
description: Executions count per namespace
legend:
enabled: true
column: namespace
data
type: io.kestra.plugin.core.dashboard.data.Executions
columns:
namespace:
field: NAMESPACE
state:
field: STATE
total:
displayName: Executions
agg: COUNT
"""
}
code = """
charts:
- id: executions_per_namespace_bars
type: io.kestra.plugin.core.dashboard.chart.Bar
chartOptions:
displayName: Executions (per namespace)
description: Executions count per namespace
legend:
enabled: true
column: namespace
data
type: io.kestra.plugin.core.dashboard.data.Executions
columns:
namespace:
field: NAMESPACE
state:
field: STATE
total:
displayName: Executions
agg: COUNT
"""
)
}
)

View File

@@ -30,26 +30,25 @@ import lombok.experimental.SuperBuilder;
@Example(
title = "Display a chart with executions in success in a given namespace.",
full = true,
code = { """
charts:
- id: kpi_success_ratio
type: io.kestra.plugin.core.dashboard.chart.KPI
chartOptions:
displayName: Success Ratio
numberType: PERCENTAGE
width: 3
data:
type: io.kestra.plugin.core.dashboard.data.ExecutionsKPI
columns:
field: ID
agg: COUNT
numerator:
- type: IN
field: STATE
values:
- SUCCESS
"""
}
code = """
charts:
- id: kpi_success_ratio
type: io.kestra.plugin.core.dashboard.chart.KPI
chartOptions:
displayName: Success Ratio
numberType: PERCENTAGE
width: 3
data:
type: io.kestra.plugin.core.dashboard.data.ExecutionsKPI
columns:
field: ID
agg: COUNT
numerator:
- type: IN
field: STATE
values:
- SUCCESS
"""
)
}
)

View File

@@ -27,19 +27,18 @@ import lombok.experimental.SuperBuilder;
@Example(
title = "Display a chart with a list of Flows.",
full = true,
code = { """
charts:
- id: list_flows
type: io.kestra.plugin.core.dashboard.chart.Table
data:
type: io.kestra.plugin.core.dashboard.data.Flows
columns:
namespace:
field: NAMESPACE
id:
field: ID
"""
}
code = """
charts:
- id: list_flows
type: io.kestra.plugin.core.dashboard.chart.Table
data:
type: io.kestra.plugin.core.dashboard.data.Flows
columns:
namespace:
field: NAMESPACE
id:
field: ID
"""
)
}
)

View File

@@ -28,17 +28,16 @@ import lombok.experimental.SuperBuilder;
@Example(
title = "Display count of Flows.",
full = true,
code = { """
charts:
- id: kpi
code = """
charts:
- id: kpi
type: io.kestra.plugin.core.dashboard.chart.KPI
data:
type: io.kestra.plugin.core.dashboard.data.FlowsKPI
columns:
field: ID
agg: COUNT
"""
}
"""
)
}
)

View File

@@ -48,11 +48,11 @@ import java.util.Optional;
id: compute_header
type: io.kestra.plugin.core.debug.Return
format: >-
{%- if inputs.token is not empty -%}
Bearer {{ inputs.token }}
{%- elseif inputs.username is not empty and inputs.password is not empty -%}
Basic {{ (inputs.username + ':' + inputs.password) | base64encode }}
{%- endif -%}
{%- if inputs.token is not empty -%}
Bearer {{ inputs.token }}
{%- elseif inputs.username is not empty and inputs.password is not empty -%}
Basic {{ (inputs.username + ':' + inputs.password) | base64encode }}
{%- endif -%}
"""
)
},

View File

@@ -54,8 +54,8 @@ import java.util.concurrent.atomic.AtomicInteger;
" - id: fail\n" +
" type: io.kestra.plugin.core.execution.Assert\n" +
" conditions:\n" +
" - \"{{ inputs.param == 'ok' }}\"\n" +
" - \"{{ 1 + 1 == 3 }}\"\n"
" - \"{{ inputs.param == 'ok' }}\"\n" +
" - \"{{ 1 + 1 == 3 }}\"\n"
}
)
},

View File

@@ -13,7 +13,6 @@ import io.kestra.core.models.tasks.Task;
import io.kestra.core.repositories.ExecutionRepositoryInterface;
import io.kestra.core.runners.DefaultRunContext;
import io.kestra.core.runners.RunContext;
import io.kestra.core.services.FlowService;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotNull;
import lombok.*;
@@ -127,14 +126,13 @@ public class Count extends Task implements RunnableTask<Count.Output> {
var flowInfo = runContext.flowInfo();
// check that all flows are allowed
FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);
if (flows != null) {
flows.forEach(flow -> flowService.checkAllowedNamespace(flowInfo.tenantId(), flow.getNamespace(), flowInfo.tenantId(), flowInfo.namespace()));
flows.forEach(flow -> runContext.acl().allowNamespace(flow.getNamespace()).check());
}
if (namespaces != null) {
var renderedNamespaces = runContext.render(this.namespaces).asList(String.class);
renderedNamespaces.forEach(namespace -> flowService.checkAllowedNamespace(flowInfo.tenantId(), namespace, flowInfo.tenantId(), flowInfo.namespace()));
renderedNamespaces.forEach(namespace -> runContext.acl().allowNamespace(namespace).check());
}
List<ExecutionCount> executionCounts = executionRepository.executionCounts(

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