mirror of
https://github.com/kestra-io/kestra.git
synced 2025-12-25 11:12:12 -05:00
Compare commits
290 Commits
docs/retur
...
plugin/tem
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0a121c6979 | ||
|
|
69a793b227 | ||
|
|
35ccb3e39b | ||
|
|
3a7fcb2aa1 | ||
|
|
103c5b92e9 | ||
|
|
5253eeef95 | ||
|
|
848f835191 | ||
|
|
3e55e67534 | ||
|
|
7bca8b4924 | ||
|
|
56febfb415 | ||
|
|
925b8c6954 | ||
|
|
708816fe67 | ||
|
|
5502473fa4 | ||
|
|
c6cf0147a4 | ||
|
|
2951f4b4bc | ||
|
|
4ea13e258b | ||
|
|
3f8dcb47fd | ||
|
|
42dc3b930c | ||
|
|
97a78abd28 | ||
|
|
b3b2ef1b5a | ||
|
|
596a26a137 | ||
|
|
8a9a1df436 | ||
|
|
55d0880ed3 | ||
|
|
a74ebd5cd6 | ||
|
|
f3aed38964 | ||
|
|
2595e56199 | ||
|
|
e821bd7f65 | ||
|
|
09762d2a8d | ||
|
|
018c22918f | ||
|
|
3e9c8cf7da | ||
|
|
008404e442 | ||
|
|
2b224bcde8 | ||
|
|
1977b61693 | ||
|
|
8e2267f86c | ||
|
|
24355c2a88 | ||
|
|
51adcfa908 | ||
|
|
a55baa1f96 | ||
|
|
32793fde18 | ||
|
|
4381d585ec | ||
|
|
e595e26c45 | ||
|
|
b833cf28b5 | ||
|
|
ac11e9545c | ||
|
|
a07df5f6cd | ||
|
|
f626c85346 | ||
|
|
e15b53ebb5 | ||
|
|
7edb6bc379 | ||
|
|
78c81f932b | ||
|
|
56bb3ca29c | ||
|
|
14029e8c14 | ||
|
|
bea3d63d89 | ||
|
|
24a3bbd303 | ||
|
|
f9932af2e8 | ||
|
|
e0410c8f24 | ||
|
|
424a6cb41a | ||
|
|
afde71e913 | ||
|
|
086c32e711 | ||
|
|
710abcfaac | ||
|
|
be951d015c | ||
|
|
a07260bef4 | ||
|
|
dd19f8391d | ||
|
|
354873e220 | ||
|
|
386d4a15f0 | ||
|
|
1b75f15680 | ||
|
|
957bf74d97 | ||
|
|
3cbad1ce0d | ||
|
|
760050e9fc | ||
|
|
43f47ec337 | ||
|
|
ad5521199a | ||
|
|
fe7849d7fe | ||
|
|
feeaeff0b2 | ||
|
|
ed6bc50163 | ||
|
|
069845f579 | ||
|
|
f613eb0433 | ||
|
|
e440c402b4 | ||
|
|
700527b5dc | ||
|
|
5245014a32 | ||
|
|
5db0f44fb6 | ||
|
|
a8635108b7 | ||
|
|
cd4470044e | ||
|
|
4ec7f23a7b | ||
|
|
107ba16ce3 | ||
|
|
042d548598 | ||
|
|
94bd6f0a1e | ||
|
|
f43f11e125 | ||
|
|
3dfa5f97c4 | ||
|
|
8898ba736b | ||
|
|
f5665bf719 | ||
|
|
b6db688003 | ||
|
|
93f5e366ed | ||
|
|
0465ffa5df | ||
|
|
e869c54883 | ||
|
|
32da15b2ea | ||
|
|
a72ecfc2eb | ||
|
|
7cb494b244 | ||
|
|
9d21ab4b26 | ||
|
|
8f3a5058b1 | ||
|
|
56fb304ff6 | ||
|
|
28370d80df | ||
|
|
cc5fd30b2c | ||
|
|
61c39d23c5 | ||
|
|
b5a40b2fcc | ||
|
|
825b9dbcdb | ||
|
|
393b132444 | ||
|
|
b0ce760e50 | ||
|
|
2d68dad70c | ||
|
|
d08a2d8930 | ||
|
|
2722735d2d | ||
|
|
e9b7d190d4 | ||
|
|
d6cfa01fd5 | ||
|
|
d8e3a9bd44 | ||
|
|
b18e3b76ef | ||
|
|
4660091dc9 | ||
|
|
067414ffbe | ||
|
|
26f6154eed | ||
|
|
ea44128d2b | ||
|
|
4602546045 | ||
|
|
aecd050314 | ||
|
|
d6c290cb91 | ||
|
|
56216ef0b4 | ||
|
|
e1f983cc2d | ||
|
|
68f92e1159 | ||
|
|
5b597b9520 | ||
|
|
b0606a4380 | ||
|
|
06450bfd65 | ||
|
|
ce12e19f99 | ||
|
|
e20d67f4f2 | ||
|
|
4d353937c3 | ||
|
|
8edad60695 | ||
|
|
c0ecc2cb20 | ||
|
|
682d258e7b | ||
|
|
d20f7039c7 | ||
|
|
4e1b53fadf | ||
|
|
2191331750 | ||
|
|
90c3281eae | ||
|
|
9fa94deba9 | ||
|
|
9d73d72ab0 | ||
|
|
4799ee320f | ||
|
|
40880bf7d8 | ||
|
|
7aca309be5 | ||
|
|
64899f3103 | ||
|
|
fbe6df34ca | ||
|
|
df21ef4064 | ||
|
|
64a2c3b746 | ||
|
|
f06b1c5347 | ||
|
|
ef154bb029 | ||
|
|
496e01eb3e | ||
|
|
f2c15185fb | ||
|
|
20c5328199 | ||
|
|
91330496f2 | ||
|
|
101700ac53 | ||
|
|
36389d7d79 | ||
|
|
f29dbe53a8 | ||
|
|
a6d34151bf | ||
|
|
4e54fac980 | ||
|
|
6e50654544 | ||
|
|
d146ebfb01 | ||
|
|
e353399d47 | ||
|
|
038083cdf4 | ||
|
|
568e66c75e | ||
|
|
5a8552ad36 | ||
|
|
da323d792a | ||
|
|
659731813a | ||
|
|
b8b20e76ba | ||
|
|
cf0b551f8f | ||
|
|
84840fe090 | ||
|
|
0dba0367f7 | ||
|
|
905341c185 | ||
|
|
b33fbc284d | ||
|
|
71f1bb9477 | ||
|
|
491e286eee | ||
|
|
469e230ebd | ||
|
|
ebb86f6d19 | ||
|
|
b68dcb7bf5 | ||
|
|
65786343ef | ||
|
|
d6933b8e49 | ||
|
|
8bd5593b2d | ||
|
|
af87713258 | ||
|
|
371c1281ca | ||
|
|
6a111a676c | ||
|
|
15da58dbf4 | ||
|
|
e37e2b0166 | ||
|
|
9f90412237 | ||
|
|
c3d94dc8ff | ||
|
|
98678deabb | ||
|
|
248c2154a2 | ||
|
|
546039e30a | ||
|
|
27bcb9c347 | ||
|
|
3f7b6a0e72 | ||
|
|
aeca59a3e4 | ||
|
|
d7caf9ae00 | ||
|
|
b5efc27763 | ||
|
|
4909978f7f | ||
|
|
f8740871ec | ||
|
|
187319ad54 | ||
|
|
9459d6556b | ||
|
|
a9d1e9ac4d | ||
|
|
c6c62dbe47 | ||
|
|
8f4bafc666 | ||
|
|
e46fbe480e | ||
|
|
7fd16b24e0 | ||
|
|
51529c8ead | ||
|
|
f53135a856 | ||
|
|
bd4eebed32 | ||
|
|
f2e7283c72 | ||
|
|
e31e833ce6 | ||
|
|
e7a99bb37f | ||
|
|
1fd4bf7499 | ||
|
|
c5851ce254 | ||
|
|
1f1976099e | ||
|
|
6a8e6b414b | ||
|
|
80d81820c9 | ||
|
|
0a62957f05 | ||
|
|
2c46bc0c39 | ||
|
|
f0189c32fc | ||
|
|
e6058f3d3e | ||
|
|
a5ec12c62a | ||
|
|
5439d395b1 | ||
|
|
bb363f8832 | ||
|
|
865aaa1fde | ||
|
|
116e5aad2d | ||
|
|
5860ce73bb | ||
|
|
527d80cd74 | ||
|
|
c99bd1d4ea | ||
|
|
c4a6ea617f | ||
|
|
a4b0beaf63 | ||
|
|
a5847aeb3a | ||
|
|
49bbc15d91 | ||
|
|
9d6694f807 | ||
|
|
eb51c5be37 | ||
|
|
90ee720d49 | ||
|
|
fd259082a6 | ||
|
|
b5323f969c | ||
|
|
6c826e93c8 | ||
|
|
aae3e6605d | ||
|
|
ea17077b0a | ||
|
|
117200eaab | ||
|
|
3216611828 | ||
|
|
1173eb2dde | ||
|
|
360b58a851 | ||
|
|
57e288abdd | ||
|
|
7fa14eb3f5 | ||
|
|
0ed2b0a53c | ||
|
|
68ace7a59b | ||
|
|
105b1b36e5 | ||
|
|
15e82f65c6 | ||
|
|
aec75bb673 | ||
|
|
f489678532 | ||
|
|
79fc5a3f24 | ||
|
|
312ec2c36b | ||
|
|
d57150e69c | ||
|
|
4b25232d4e | ||
|
|
1d1a065833 | ||
|
|
d6ecbadee1 | ||
|
|
205605060d | ||
|
|
6ef3a00e16 | ||
|
|
f70d612878 | ||
|
|
0b345c03d1 | ||
|
|
ecb508f797 | ||
|
|
38caea2568 | ||
|
|
cdad732576 | ||
|
|
0dd4cb963f | ||
|
|
c35ca82356 | ||
|
|
db6cb93df4 | ||
|
|
1f8d2ea918 | ||
|
|
3c09a38eed | ||
|
|
0525e3ece6 | ||
|
|
4382aabe39 | ||
|
|
5e0fddadc4 | ||
|
|
4297459a6a | ||
|
|
31e5f6bdef | ||
|
|
dc7cea0396 | ||
|
|
e818614f4a | ||
|
|
a5ccfbb0ac | ||
|
|
e5fece8d4d | ||
|
|
816a1bb543 | ||
|
|
735697ac71 | ||
|
|
4fc6948037 | ||
|
|
e56e35e770 | ||
|
|
a4ca3498f3 | ||
|
|
d7e17f157a | ||
|
|
41f83949f0 | ||
|
|
0db2d8759a | ||
|
|
6e0197b542 | ||
|
|
6918d5d512 | ||
|
|
a3fc9b1532 | ||
|
|
0f340a9a29 | ||
|
|
8a8911a25d | ||
|
|
ae204a03b0 | ||
|
|
4dc7924184 | ||
|
|
748d055183 |
1
.github/ISSUE_TEMPLATE/bug.yml
vendored
1
.github/ISSUE_TEMPLATE/bug.yml
vendored
@@ -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
|
||||
|
||||
1
.github/ISSUE_TEMPLATE/feature.yml
vendored
1
.github/ISSUE_TEMPLATE/feature.yml
vendored
@@ -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
|
||||
|
||||
95
.github/dependabot.yml
vendored
95
.github/dependabot.yml
vendored
@@ -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"]
|
||||
|
||||
2
.github/workflows/auto-translate-ui-keys.yml
vendored
2
.github/workflows/auto-translate-ui-keys.yml
vendored
@@ -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
|
||||
|
||||
2
.github/workflows/codeql-analysis.yml
vendored
2
.github/workflows/codeql-analysis.yml
vendored
@@ -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.
|
||||
|
||||
@@ -33,7 +33,7 @@ jobs:
|
||||
exit 1;
|
||||
fi
|
||||
# Checkout
|
||||
- uses: actions/checkout@v5
|
||||
- uses: actions/checkout@v6
|
||||
with:
|
||||
fetch-depth: 0
|
||||
path: kestra
|
||||
|
||||
2
.github/workflows/global-start-release.yml
vendored
2
.github/workflows/global-start-release.yml
vendored
@@ -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 }}
|
||||
|
||||
3
.github/workflows/main-build.yml
vendored
3
.github/workflows/main-build.yml
vendored
@@ -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:
|
||||
|
||||
6
.github/workflows/pull-request.yml
vendored
6
.github/workflows/pull-request.yml
vendored
@@ -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 }}
|
||||
|
||||
1
.github/workflows/release-docker.yml
vendored
1
.github/workflows/release-docker.yml
vendored
@@ -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 }}
|
||||
6
.github/workflows/vulnerabilities-check.yml
vendored
6
.github/workflows/vulnerabilities-check.yml
vendored
@@ -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
|
||||
|
||||
|
||||
20
build.gradle
20
build.gradle
@@ -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 ->
|
||||
|
||||
@@ -8,11 +8,10 @@ import io.kestra.cli.commands.plugins.PluginCommand;
|
||||
import io.kestra.cli.commands.servers.ServerCommand;
|
||||
import io.kestra.cli.commands.sys.SysCommand;
|
||||
import io.kestra.cli.commands.templates.TemplateCommand;
|
||||
import io.kestra.cli.services.EnvironmentProvider;
|
||||
import io.micronaut.configuration.picocli.MicronautFactory;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import io.micronaut.context.ApplicationContext;
|
||||
import io.micronaut.context.ApplicationContextBuilder;
|
||||
import io.micronaut.context.env.Environment;
|
||||
import io.micronaut.core.annotation.Introspected;
|
||||
import org.slf4j.bridge.SLF4JBridgeHandler;
|
||||
import picocli.CommandLine;
|
||||
@@ -20,11 +19,9 @@ import picocli.CommandLine;
|
||||
import java.lang.reflect.Field;
|
||||
import java.lang.reflect.InvocationTargetException;
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
@CommandLine.Command(
|
||||
name = "kestra",
|
||||
@@ -49,35 +46,77 @@ import java.util.concurrent.Callable;
|
||||
@Introspected
|
||||
public class App implements Callable<Integer> {
|
||||
public static void main(String[] args) {
|
||||
execute(App.class, new String [] { Environment.CLI }, args);
|
||||
System.exit(runCli(args));
|
||||
}
|
||||
|
||||
public static int runCli(String[] args, String... extraEnvironments) {
|
||||
return runCli(App.class, args, extraEnvironments);
|
||||
}
|
||||
|
||||
public static int runCli(Class<?> cls, String[] args, String... extraEnvironments) {
|
||||
ServiceLoader<EnvironmentProvider> environmentProviders = ServiceLoader.load(EnvironmentProvider.class);
|
||||
String[] baseEnvironments = environmentProviders.findFirst().map(EnvironmentProvider::getCliEnvironments).orElseGet(() -> new String[0]);
|
||||
return execute(
|
||||
cls,
|
||||
Stream.concat(
|
||||
Arrays.stream(baseEnvironments),
|
||||
Arrays.stream(extraEnvironments)
|
||||
).toArray(String[]::new),
|
||||
args
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Integer call() throws Exception {
|
||||
return PicocliRunner.call(App.class, "--help");
|
||||
return runCli(new String[0]);
|
||||
}
|
||||
|
||||
protected static void execute(Class<?> cls, String[] environments, String... args) {
|
||||
protected static int execute(Class<?> cls, String[] environments, String... args) {
|
||||
// Log Bridge
|
||||
SLF4JBridgeHandler.removeHandlersForRootLogger();
|
||||
SLF4JBridgeHandler.install();
|
||||
|
||||
// Init ApplicationContext
|
||||
ApplicationContext applicationContext = App.applicationContext(cls, environments, args);
|
||||
CommandLine commandLine = getCommandLine(cls, args);
|
||||
|
||||
ApplicationContext applicationContext = App.applicationContext(cls, commandLine, environments);
|
||||
|
||||
Class<?> targetCommand = commandLine.getCommandSpec().userObject().getClass();
|
||||
|
||||
if (!AbstractCommand.class.isAssignableFrom(targetCommand) && args.length == 0) {
|
||||
// if no command provided, show help
|
||||
args = new String[]{"--help"};
|
||||
}
|
||||
|
||||
// Call Picocli command
|
||||
int exitCode = 0;
|
||||
int exitCode;
|
||||
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;
|
||||
}
|
||||
applicationContext.close();
|
||||
|
||||
// exit code
|
||||
System.exit(Objects.requireNonNullElse(exitCode, 0));
|
||||
return exitCode;
|
||||
}
|
||||
|
||||
private static CommandLine getCommandLine(Class<?> cls, String[] args) {
|
||||
CommandLine cmd = new CommandLine(cls, CommandLine.defaultFactory());
|
||||
continueOnParsingErrors(cmd);
|
||||
|
||||
CommandLine.ParseResult parseResult = cmd.parseArgs(args);
|
||||
List<CommandLine> parsedCommands = parseResult.asCommandLineList();
|
||||
|
||||
return parsedCommands.getLast();
|
||||
}
|
||||
|
||||
public static ApplicationContext applicationContext(Class<?> mainClass,
|
||||
String[] environments,
|
||||
String... args) {
|
||||
return App.applicationContext(mainClass, getCommandLine(mainClass, args), environments);
|
||||
}
|
||||
|
||||
|
||||
@@ -85,25 +124,17 @@ public class App implements Callable<Integer> {
|
||||
* Create an {@link ApplicationContext} with additional properties based on configuration files (--config) and
|
||||
* forced Properties from current command.
|
||||
*
|
||||
* @param args args passed to java app
|
||||
* @return the application context created
|
||||
*/
|
||||
protected static ApplicationContext applicationContext(Class<?> mainClass,
|
||||
String[] environments,
|
||||
String[] args) {
|
||||
CommandLine commandLine,
|
||||
String[] environments) {
|
||||
|
||||
ApplicationContextBuilder builder = ApplicationContext
|
||||
.builder()
|
||||
.mainClass(mainClass)
|
||||
.environments(environments);
|
||||
|
||||
CommandLine cmd = new CommandLine(mainClass, CommandLine.defaultFactory());
|
||||
continueOnParsingErrors(cmd);
|
||||
|
||||
CommandLine.ParseResult parseResult = cmd.parseArgs(args);
|
||||
List<CommandLine> parsedCommands = parseResult.asCommandLineList();
|
||||
|
||||
CommandLine commandLine = parsedCommands.getLast();
|
||||
Class<?> cls = commandLine.getCommandSpec().userObject().getClass();
|
||||
|
||||
if (AbstractCommand.class.isAssignableFrom(cls)) {
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
package io.kestra.cli.commands.configs.sys;
|
||||
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
@@ -20,8 +19,6 @@ public class ConfigCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "configs", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"configs", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
package io.kestra.cli.commands.flows;
|
||||
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
@@ -29,8 +28,6 @@ public class FlowCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "flow", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"flow", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package io.kestra.cli.commands.flows.namespaces;
|
||||
|
||||
import io.kestra.cli.App;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
@@ -22,8 +21,6 @@ public class FlowNamespaceCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "flow", "namespace", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"flow", "namespace", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package io.kestra.cli.commands.migrations;
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.kestra.cli.commands.migrations.metadata.MetadataMigrationCommand;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import picocli.CommandLine;
|
||||
@@ -24,8 +23,6 @@ public class MigrationCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "migrate", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"migrate", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,7 +10,8 @@ import picocli.CommandLine;
|
||||
description = "populate metadata for entities",
|
||||
subcommands = {
|
||||
KvMetadataMigrationCommand.class,
|
||||
SecretsMetadataMigrationCommand.class
|
||||
SecretsMetadataMigrationCommand.class,
|
||||
NsFilesMetadataMigrationCommand.class
|
||||
}
|
||||
)
|
||||
@Slf4j
|
||||
|
||||
@@ -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) {}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,6 @@ import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.kestra.cli.commands.namespaces.files.NamespaceFilesCommand;
|
||||
import io.kestra.cli.commands.namespaces.kv.KvCommand;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import picocli.CommandLine;
|
||||
@@ -25,8 +24,6 @@ public class NamespaceCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "namespace", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"namespace", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package io.kestra.cli.commands.namespaces.files;
|
||||
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import picocli.CommandLine;
|
||||
@@ -22,8 +21,6 @@ public class NamespaceFilesCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "namespace", "files", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"namespace", "files", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package io.kestra.cli.commands.namespaces.kv;
|
||||
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import picocli.CommandLine;
|
||||
@@ -22,8 +21,6 @@ public class KvCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "namespace", "kv", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"namespace", "kv", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package io.kestra.cli.commands.plugins;
|
||||
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import picocli.CommandLine.Command;
|
||||
|
||||
@@ -25,9 +24,7 @@ public class PluginCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "plugins", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"plugins", "--help"});
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
package io.kestra.cli.commands.servers;
|
||||
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
@@ -28,8 +27,6 @@ public class ServerCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "server", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"server", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package io.kestra.cli.commands.sys;
|
||||
|
||||
import io.kestra.cli.commands.sys.database.DatabaseCommand;
|
||||
import io.kestra.cli.commands.sys.statestore.StateStoreCommand;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
@@ -25,8 +24,6 @@ public class SysCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "sys", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"sys", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package io.kestra.cli.commands.sys.database;
|
||||
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import picocli.CommandLine;
|
||||
|
||||
@@ -20,8 +19,6 @@ public class DatabaseCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "sys", "database", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"sys", "database", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package io.kestra.cli.commands.sys.statestore;
|
||||
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import picocli.CommandLine;
|
||||
|
||||
@@ -20,8 +19,6 @@ public class StateStoreCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "sys", "state-store", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"sys", "state-store", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.kestra.cli.commands.templates.namespaces.TemplateNamespaceCommand;
|
||||
import io.kestra.core.models.templates.TemplateEnabled;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import picocli.CommandLine;
|
||||
@@ -27,8 +26,6 @@ public class TemplateCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "template", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"template", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package io.kestra.cli.commands.templates.namespaces;
|
||||
import io.kestra.cli.AbstractCommand;
|
||||
import io.kestra.cli.App;
|
||||
import io.kestra.core.models.templates.TemplateEnabled;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import lombok.SneakyThrows;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import picocli.CommandLine;
|
||||
@@ -24,8 +23,6 @@ public class TemplateNamespaceCommand extends AbstractCommand {
|
||||
public Integer call() throws Exception {
|
||||
super.call();
|
||||
|
||||
PicocliRunner.call(App.class, "template", "namespace", "--help");
|
||||
|
||||
return 0;
|
||||
return App.runCli(new String[]{"template", "namespace", "--help"});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
package io.kestra.cli.services;
|
||||
|
||||
import io.micronaut.context.env.Environment;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
public class DefaultEnvironmentProvider implements EnvironmentProvider {
|
||||
@Override
|
||||
public String[] getCliEnvironments(String... extraEnvironments) {
|
||||
return Stream.concat(
|
||||
Stream.of(Environment.CLI),
|
||||
Arrays.stream(extraEnvironments)
|
||||
).toArray(String[]::new);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
package io.kestra.cli.services;
|
||||
|
||||
public interface EnvironmentProvider {
|
||||
String[] getCliEnvironments(String... extraEnvironments);
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
io.kestra.cli.services.DefaultEnvironmentProvider
|
||||
@@ -1,14 +1,11 @@
|
||||
package io.kestra.cli;
|
||||
|
||||
import io.kestra.core.models.ServerType;
|
||||
import io.micronaut.configuration.picocli.MicronautFactory;
|
||||
import io.micronaut.configuration.picocli.PicocliRunner;
|
||||
import io.micronaut.context.ApplicationContext;
|
||||
import io.micronaut.context.env.Environment;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import picocli.CommandLine;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.PrintStream;
|
||||
@@ -22,11 +19,15 @@ class AppTest {
|
||||
ByteArrayOutputStream out = new ByteArrayOutputStream();
|
||||
System.setOut(new PrintStream(out));
|
||||
|
||||
try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
|
||||
PicocliRunner.call(App.class, ctx, "--help");
|
||||
// No arg will print help
|
||||
assertThat(App.runCli(new String[0])).isZero();
|
||||
assertThat(out.toString()).contains("kestra");
|
||||
|
||||
assertThat(out.toString()).contains("kestra");
|
||||
}
|
||||
out.reset();
|
||||
|
||||
// Explicit help command
|
||||
assertThat(App.runCli(new String[]{"--help"})).isZero();
|
||||
assertThat(out.toString()).contains("kestra");
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@@ -38,11 +39,12 @@ class AppTest {
|
||||
final String[] args = new String[]{"server", serverType, "--help"};
|
||||
|
||||
try (ApplicationContext ctx = App.applicationContext(App.class, new String [] { Environment.CLI }, args)) {
|
||||
new CommandLine(App.class, new MicronautFactory(ctx)).execute(args);
|
||||
|
||||
assertTrue(ctx.getProperty("kestra.server-type", ServerType.class).isEmpty());
|
||||
assertThat(out.toString()).startsWith("Usage: kestra server " + serverType);
|
||||
}
|
||||
|
||||
assertThat(App.runCli(args)).isZero();
|
||||
|
||||
assertThat(out.toString()).startsWith("Usage: kestra server " + serverType);
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -52,12 +54,10 @@ class AppTest {
|
||||
|
||||
final String[] argsWithMissingParams = new String[]{"flow", "namespace", "update"};
|
||||
|
||||
try (ApplicationContext ctx = App.applicationContext(App.class, new String [] { Environment.CLI }, argsWithMissingParams)) {
|
||||
new CommandLine(App.class, new MicronautFactory(ctx)).execute(argsWithMissingParams);
|
||||
assertThat(App.runCli(argsWithMissingParams)).isEqualTo(2);
|
||||
|
||||
assertThat(out.toString()).startsWith("Missing required parameters: ");
|
||||
assertThat(out.toString()).contains("Usage: kestra flow namespace update ");
|
||||
assertThat(out.toString()).doesNotContain("MissingParameterException: ");
|
||||
}
|
||||
assertThat(out.toString()).startsWith("Missing required parameters: ");
|
||||
assertThat(out.toString()).contains("Usage: kestra flow namespace update ");
|
||||
assertThat(out.toString()).doesNotContain("MissingParameterException: ");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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"));
|
||||
|
||||
|
||||
@@ -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)
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ kestra:
|
||||
server:
|
||||
liveness:
|
||||
enabled: false
|
||||
termination-grace-period: 5s
|
||||
micronaut:
|
||||
http:
|
||||
services:
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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));
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -3,7 +3,6 @@ package io.kestra.core.models.conditions;
|
||||
import io.kestra.core.models.flows.FlowInterface;
|
||||
import lombok.*;
|
||||
import io.kestra.core.models.executions.Execution;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
|
||||
import io.kestra.core.runners.RunContext;
|
||||
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -3,7 +3,7 @@ package io.kestra.core.models.executions;
|
||||
import com.fasterxml.jackson.annotation.JsonInclude;
|
||||
import io.kestra.core.models.DeletedInterface;
|
||||
import io.kestra.core.models.TenantInterface;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.flows.FlowInterface;
|
||||
import io.kestra.core.models.triggers.AbstractTrigger;
|
||||
import io.kestra.core.models.triggers.TriggerContext;
|
||||
import io.swagger.v3.oas.annotations.Hidden;
|
||||
@@ -97,7 +97,7 @@ public class LogEntry implements DeletedInterface, TenantInterface {
|
||||
.build();
|
||||
}
|
||||
|
||||
public static LogEntry of(Flow flow, AbstractTrigger abstractTrigger, ExecutionKind executionKind) {
|
||||
public static LogEntry of(FlowInterface flow, AbstractTrigger abstractTrigger) {
|
||||
return LogEntry.builder()
|
||||
.tenantId(flow.getTenantId())
|
||||
.namespace(flow.getNamespace())
|
||||
@@ -107,7 +107,7 @@ public class LogEntry implements DeletedInterface, TenantInterface {
|
||||
.build();
|
||||
}
|
||||
|
||||
public static LogEntry of(TriggerContext triggerContext, AbstractTrigger abstractTrigger, ExecutionKind executionKind) {
|
||||
public static LogEntry of(TriggerContext triggerContext, AbstractTrigger abstractTrigger) {
|
||||
return LogEntry.builder()
|
||||
.tenantId(triggerContext.getTenantId())
|
||||
.namespace(triggerContext.getNamespace())
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
109
core/src/main/java/io/kestra/core/models/flows/check/Check.java
Normal file
109
core/src/main/java/io/kestra/core/models/flows/check/Check.java
Normal 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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -48,7 +48,7 @@ public class SubflowGraphTask extends AbstractGraphTask {
|
||||
|
||||
public record SubflowTaskWrapper<T extends Output>(RunContext runContext, ExecutableTask<T> subflowTask) implements TaskInterface, ExecutableTask<T> {
|
||||
@Override
|
||||
public List<SubflowExecution<?>> createSubflowExecutions(RunContext runContext, FlowMetaStoreInterface flowExecutorInterface, Flow currentFlow, Execution currentExecution, TaskRun currentTaskRun) throws InternalException {
|
||||
public List<SubflowExecution<?>> createSubflowExecutions(RunContext runContext, FlowMetaStoreInterface flowExecutorInterface, FlowInterface currentFlow, Execution currentExecution, TaskRun currentTaskRun) throws InternalException {
|
||||
return subflowTask.createSubflowExecutions(runContext, flowExecutorInterface, currentFlow, currentExecution, currentTaskRun);
|
||||
}
|
||||
|
||||
|
||||
@@ -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()));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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("/");
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -24,7 +24,7 @@ public interface ExecutableTask<T extends Output>{
|
||||
*/
|
||||
List<SubflowExecution<?>> createSubflowExecutions(RunContext runContext,
|
||||
FlowMetaStoreInterface flowExecutorInterface,
|
||||
Flow currentFlow, Execution currentExecution,
|
||||
FlowInterface currentFlow, Execution currentExecution,
|
||||
TaskRun currentTaskRun) throws InternalException;
|
||||
|
||||
/**
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -74,7 +74,7 @@ public class Trigger extends TriggerContext implements HasUID {
|
||||
);
|
||||
}
|
||||
|
||||
public static String uid(Flow flow, AbstractTrigger abstractTrigger) {
|
||||
public static String uid(FlowInterface flow, AbstractTrigger abstractTrigger) {
|
||||
return IdUtils.fromParts(
|
||||
flow.getTenantId(),
|
||||
flow.getNamespace(),
|
||||
|
||||
@@ -2,14 +2,12 @@ package io.kestra.core.models.triggers.multipleflows;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import io.kestra.core.models.HasUID;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.flows.FlowId;
|
||||
import io.kestra.core.utils.IdUtils;
|
||||
import lombok.Builder;
|
||||
import lombok.Value;
|
||||
|
||||
import java.time.ZonedDateTime;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
|
||||
@@ -23,12 +23,12 @@ import java.util.Objects;
|
||||
|
||||
@Singleton
|
||||
public class FeatureUsageReport extends AbstractReportable<FeatureUsageReport.UsageEvent> {
|
||||
|
||||
|
||||
private final FlowRepositoryInterface flowRepository;
|
||||
private final ExecutionRepositoryInterface executionRepository;
|
||||
private final DashboardRepositoryInterface dashboardRepository;
|
||||
private final boolean enabled;
|
||||
|
||||
|
||||
@Inject
|
||||
public FeatureUsageReport(FlowRepositoryInterface flowRepository,
|
||||
ExecutionRepositoryInterface executionRepository,
|
||||
@@ -37,26 +37,26 @@ public class FeatureUsageReport extends AbstractReportable<FeatureUsageReport.Us
|
||||
this.flowRepository = flowRepository;
|
||||
this.executionRepository = executionRepository;
|
||||
this.dashboardRepository = dashboardRepository;
|
||||
|
||||
|
||||
ServerType serverType = KestraContext.getContext().getServerType();
|
||||
this.enabled = ServerType.EXECUTOR.equals(serverType) || ServerType.STANDALONE.equals(serverType);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public UsageEvent report(final Instant now, TimeInterval interval) {
|
||||
return UsageEvent
|
||||
.builder()
|
||||
.flows(FlowUsage.of(flowRepository))
|
||||
.executions(ExecutionUsage.of(executionRepository, interval.from(), interval.to()))
|
||||
.dashboards(new Count(dashboardRepository.count()))
|
||||
.dashboards(new Count(dashboardRepository.countAllForAllTenants()))
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public boolean isEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public UsageEvent report(Instant now, TimeInterval interval, String tenant) {
|
||||
Objects.requireNonNull(tenant, "tenant is null");
|
||||
@@ -67,7 +67,7 @@ public class FeatureUsageReport extends AbstractReportable<FeatureUsageReport.Us
|
||||
.executions(ExecutionUsage.of(tenant, executionRepository, interval.from(), interval.to()))
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
@SuperBuilder(toBuilder = true)
|
||||
@Getter
|
||||
@Jacksonized
|
||||
|
||||
@@ -16,14 +16,14 @@ import java.util.Map;
|
||||
import java.util.Optional;
|
||||
|
||||
public interface DashboardRepositoryInterface {
|
||||
|
||||
|
||||
/**
|
||||
* Gets the total number of Dashboards.
|
||||
*
|
||||
* @return the total number.
|
||||
*/
|
||||
long count();
|
||||
|
||||
long countAllForAllTenants();
|
||||
|
||||
Boolean isEnabled();
|
||||
|
||||
Optional<Dashboard> get(String tenantId, String id);
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
@@ -39,13 +39,13 @@ public interface TriggerRepositoryInterface extends QueryBuilderInterface<Trigge
|
||||
* @param tenantId the tenant of the triggers
|
||||
* @return The count.
|
||||
*/
|
||||
int count(@Nullable String tenantId);
|
||||
long countAll(@Nullable String tenantId);
|
||||
|
||||
/**
|
||||
* 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();
|
||||
|
||||
50
core/src/main/java/io/kestra/core/runners/AclChecker.java
Normal file
50
core/src/main/java/io/kestra/core/runners/AclChecker.java
Normal 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();
|
||||
}
|
||||
}
|
||||
@@ -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()));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -26,7 +26,6 @@ import org.apache.commons.lang3.stream.Streams;
|
||||
import java.time.Instant;
|
||||
import java.time.ZonedDateTime;
|
||||
import java.util.*;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static io.kestra.core.trace.Tracer.throwCallable;
|
||||
import static io.kestra.core.utils.Rethrow.throwConsumer;
|
||||
@@ -54,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();
|
||||
}
|
||||
|
||||
@@ -67,7 +64,7 @@ public final class ExecutableUtils {
|
||||
RunContext runContext,
|
||||
FlowMetaStoreInterface flowExecutorInterface,
|
||||
Execution currentExecution,
|
||||
Flow currentFlow,
|
||||
FlowInterface currentFlow,
|
||||
T currentTask,
|
||||
TaskRun currentTaskRun,
|
||||
Map<String, Object> inputs,
|
||||
|
||||
@@ -7,7 +7,6 @@ import io.kestra.core.exceptions.KestraRuntimeException;
|
||||
import io.kestra.core.models.executions.Execution;
|
||||
import io.kestra.core.models.flows.Data;
|
||||
import io.kestra.core.models.flows.DependsOn;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.flows.FlowInterface;
|
||||
import io.kestra.core.models.flows.Input;
|
||||
import io.kestra.core.models.flows.Output;
|
||||
@@ -64,11 +63,11 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
|
||||
public class FlowInputOutput {
|
||||
private static final Pattern URI_PATTERN = Pattern.compile("^[a-z]+:\\/\\/(?:www\\.)?[-a-zA-Z0-9@:%._\\+~#=]{1,256}\\.[a-zA-Z0-9()]{1,6}\\b(?:[-a-zA-Z0-9()@:%_\\+.~#?&\\/=]*)$");
|
||||
private static final ObjectMapper YAML_MAPPER = JacksonMapper.ofYaml();
|
||||
|
||||
|
||||
private final StorageInterface storageInterface;
|
||||
private final Optional<String> secretKey;
|
||||
private final RunContextFactory runContextFactory;
|
||||
|
||||
|
||||
@Inject
|
||||
public FlowInputOutput(
|
||||
StorageInterface storageInterface,
|
||||
@@ -79,7 +78,7 @@ public class FlowInputOutput {
|
||||
this.runContextFactory = runContextFactory;
|
||||
this.secretKey = Optional.ofNullable(secretKey);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Validate all the inputs of a given execution of a flow.
|
||||
*
|
||||
@@ -89,15 +88,15 @@ public class FlowInputOutput {
|
||||
* @return The list of {@link InputAndValue}.
|
||||
*/
|
||||
public Mono<List<InputAndValue>> validateExecutionInputs(final List<Input<?>> inputs,
|
||||
final Flow flow,
|
||||
final FlowInterface flow,
|
||||
final Execution execution,
|
||||
final Publisher<CompletedPart> data) {
|
||||
if (ListUtils.isEmpty(inputs)) return Mono.just(Collections.emptyList());
|
||||
|
||||
|
||||
return readData(inputs, execution, data, false)
|
||||
.map(inputData -> resolveInputs(inputs, flow, execution, inputData, false));
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Reads all the inputs of a given execution of a flow.
|
||||
*
|
||||
@@ -111,7 +110,7 @@ public class FlowInputOutput {
|
||||
final Publisher<CompletedPart> data) {
|
||||
return this.readExecutionInputs(flow.getInputs(), flow, execution, data);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Reads all the inputs of a given execution of a flow.
|
||||
*
|
||||
@@ -126,7 +125,7 @@ public class FlowInputOutput {
|
||||
final Publisher<CompletedPart> data) {
|
||||
return readData(inputs, execution, data, true).map(inputData -> this.readExecutionInputs(inputs, flow, execution, inputData));
|
||||
}
|
||||
|
||||
|
||||
private Mono<Map<String, Object>> readData(List<Input<?>> inputs, Execution execution, Publisher<CompletedPart> data, boolean uploadFiles) {
|
||||
return Flux.from(data)
|
||||
.publishOn(Schedulers.boundedElastic())
|
||||
@@ -159,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 {
|
||||
@@ -235,7 +230,7 @@ public class FlowInputOutput {
|
||||
}
|
||||
return MapUtils.flattenToNestedMap(resolved);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Utility method for retrieving types inputs.
|
||||
*
|
||||
@@ -252,7 +247,7 @@ public class FlowInputOutput {
|
||||
) {
|
||||
return resolveInputs(inputs, flow, execution, data, true);
|
||||
}
|
||||
|
||||
|
||||
public List<InputAndValue> resolveInputs(
|
||||
final List<Input<?>> inputs,
|
||||
final FlowInterface flow,
|
||||
@@ -325,7 +320,7 @@ public class FlowInputOutput {
|
||||
}
|
||||
});
|
||||
resolvable.setInput(input);
|
||||
|
||||
|
||||
Object value = resolvable.get().value();
|
||||
|
||||
// resolve default if needed
|
||||
@@ -383,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) {
|
||||
@@ -503,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()) {
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -6,16 +6,16 @@ import com.google.common.annotations.VisibleForTesting;
|
||||
import io.kestra.core.metrics.MetricRegistry;
|
||||
import io.kestra.core.models.executions.Execution;
|
||||
import io.kestra.core.models.executions.TaskRun;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.flows.FlowInterface;
|
||||
import io.kestra.core.models.flows.Type;
|
||||
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;
|
||||
@@ -41,7 +41,7 @@ public class RunContextFactory {
|
||||
|
||||
@Inject
|
||||
protected VariableRenderer variableRenderer;
|
||||
|
||||
|
||||
@Inject
|
||||
protected SecureVariableRendererFactory secureVariableRendererFactory;
|
||||
|
||||
@@ -49,7 +49,7 @@ public class RunContextFactory {
|
||||
protected StorageInterface storageInterface;
|
||||
|
||||
@Inject
|
||||
protected FlowService flowService;
|
||||
protected NamespaceService namespaceService;
|
||||
|
||||
@Inject
|
||||
protected MetricRegistry metricRegistry;
|
||||
@@ -77,15 +77,18 @@ public class RunContextFactory {
|
||||
@Inject
|
||||
private KVStoreService kvStoreService;
|
||||
|
||||
@Inject
|
||||
private NamespaceFactory namespaceFactory;
|
||||
|
||||
// hacky
|
||||
public RunContextInitializer initializer() {
|
||||
return applicationContext.getBean(RunContextInitializer.class);
|
||||
}
|
||||
|
||||
|
||||
public RunContext of(FlowInterface flow, Execution execution) {
|
||||
return of(flow, execution, Function.identity());
|
||||
}
|
||||
|
||||
|
||||
public RunContext of(FlowInterface flow, Execution execution, boolean decryptVariable) {
|
||||
return of(flow, execution, Function.identity(), decryptVariable);
|
||||
}
|
||||
@@ -93,18 +96,18 @@ public class RunContextFactory {
|
||||
public RunContext of(FlowInterface flow, Execution execution, Function<RunVariables.Builder, RunVariables.Builder> runVariableModifier) {
|
||||
return of(flow, execution, runVariableModifier, true);
|
||||
}
|
||||
|
||||
|
||||
public RunContext of(FlowInterface flow, Execution execution, Function<RunVariables.Builder, RunVariables.Builder> runVariableModifier, boolean decryptVariables) {
|
||||
RunContextLogger runContextLogger = runContextLoggerFactory.create(execution);
|
||||
|
||||
|
||||
VariableRenderer variableRenderer = decryptVariables ? this.variableRenderer : secureVariableRendererFactory.createOrGet();
|
||||
|
||||
|
||||
return newBuilder()
|
||||
// Logger
|
||||
.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()
|
||||
@@ -134,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)
|
||||
@@ -150,8 +153,8 @@ public class RunContextFactory {
|
||||
.build();
|
||||
}
|
||||
|
||||
public RunContext of(Flow flow, AbstractTrigger trigger) {
|
||||
RunContextLogger runContextLogger = runContextLoggerFactory.create(flow, trigger, null);
|
||||
public RunContext of(FlowInterface flow, AbstractTrigger trigger) {
|
||||
RunContextLogger runContextLogger = runContextLoggerFactory.create(flow, trigger);
|
||||
return newBuilder()
|
||||
// Logger
|
||||
.withLogger(runContextLogger)
|
||||
@@ -168,14 +171,16 @@ public class RunContextFactory {
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
@VisibleForTesting
|
||||
public RunContext of(final Flow flow, final Map<String, Object> variables) {
|
||||
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();
|
||||
}
|
||||
@@ -213,7 +218,8 @@ public class RunContextFactory {
|
||||
}
|
||||
},
|
||||
storageInterface,
|
||||
flowService
|
||||
namespaceService,
|
||||
namespaceFactory
|
||||
))
|
||||
.withVariables(variables)
|
||||
.withTask(task)
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -213,7 +217,7 @@ public class RunContextInitializer {
|
||||
runContext.init(applicationContext);
|
||||
|
||||
final String triggerExecutionId = IdUtils.create();
|
||||
final RunContextLogger runContextLogger = contextLoggerFactory.create(triggerContext, trigger, null);
|
||||
final RunContextLogger runContextLogger = contextLoggerFactory.create(triggerContext, trigger);
|
||||
|
||||
final Map<String, Object> variables = new HashMap<>(runContext.getVariables());
|
||||
variables.put(RunVariables.SECRET_CONSUMER_VARIABLE_NAME, (Consumer<String>) runContextLogger::usedSecret);
|
||||
@@ -230,7 +234,8 @@ public class RunContextInitializer {
|
||||
runContextLogger.logger(),
|
||||
context,
|
||||
storageInterface,
|
||||
flowService
|
||||
namespaceService,
|
||||
namespaceFactory
|
||||
);
|
||||
|
||||
runContext.setLogger(runContextLogger);
|
||||
|
||||
@@ -4,7 +4,7 @@ import io.kestra.core.models.executions.Execution;
|
||||
import io.kestra.core.models.executions.ExecutionKind;
|
||||
import io.kestra.core.models.executions.LogEntry;
|
||||
import io.kestra.core.models.executions.TaskRun;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.flows.FlowInterface;
|
||||
import io.kestra.core.models.tasks.Task;
|
||||
import io.kestra.core.models.triggers.AbstractTrigger;
|
||||
import io.kestra.core.models.triggers.TriggerContext;
|
||||
@@ -46,19 +46,19 @@ public class RunContextLoggerFactory {
|
||||
);
|
||||
}
|
||||
|
||||
public RunContextLogger create(TriggerContext triggerContext, AbstractTrigger trigger, ExecutionKind executionKind) {
|
||||
public RunContextLogger create(TriggerContext triggerContext, AbstractTrigger trigger) {
|
||||
return new RunContextLogger(
|
||||
logQueue,
|
||||
LogEntry.of(triggerContext, trigger, executionKind),
|
||||
LogEntry.of(triggerContext, trigger),
|
||||
trigger.getLogLevel(),
|
||||
trigger.isLogToFile()
|
||||
);
|
||||
}
|
||||
|
||||
public RunContextLogger create(Flow flow, AbstractTrigger trigger, ExecutionKind executionKind) {
|
||||
public RunContextLogger create(FlowInterface flow, AbstractTrigger trigger) {
|
||||
return new RunContextLogger(
|
||||
logQueue,
|
||||
LogEntry.of(flow, trigger, executionKind),
|
||||
LogEntry.of(flow, trigger),
|
||||
trigger.getLogLevel(),
|
||||
trigger.isLogToFile()
|
||||
);
|
||||
|
||||
@@ -5,8 +5,8 @@ import io.kestra.core.exceptions.IllegalVariableEvaluationException;
|
||||
import io.kestra.core.models.Label;
|
||||
import io.kestra.core.models.executions.Execution;
|
||||
import io.kestra.core.models.executions.TaskRun;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.flows.FlowInterface;
|
||||
import io.kestra.core.models.flows.GenericFlow;
|
||||
import io.kestra.core.models.flows.Input;
|
||||
import io.kestra.core.models.flows.State;
|
||||
import io.kestra.core.models.flows.input.SecretInput;
|
||||
@@ -73,7 +73,7 @@ public final class RunVariables {
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an immutable map representation of the given {@link Flow}.
|
||||
* Creates an immutable map representation of the given {@link FlowInterface}.
|
||||
*
|
||||
* @param flow The flow from which to create variables.
|
||||
* @return a new immutable {@link Map}.
|
||||
@@ -283,7 +283,7 @@ public final class RunVariables {
|
||||
if (flow != null && flow.getInputs() != null) {
|
||||
// Create a new PropertyContext with 'flow' variables which are required by some pebble expressions.
|
||||
PropertyContextWithVariables context = new PropertyContextWithVariables(propertyContext, Map.of("flow", RunVariables.of(flow)));
|
||||
|
||||
|
||||
// we add default inputs value from the flow if not already set, this will be useful for triggers
|
||||
flow.getInputs().stream()
|
||||
.filter(input -> input.getDefaults() != null && !inputs.containsKey(input.getId()))
|
||||
@@ -326,7 +326,7 @@ public final class RunVariables {
|
||||
}
|
||||
|
||||
if (flow == null) {
|
||||
Flow flowFromExecution = Flow.builder()
|
||||
FlowInterface flowFromExecution = GenericFlow.builder()
|
||||
.id(execution.getFlowId())
|
||||
.tenantId(execution.getTenantId())
|
||||
.revision(execution.getFlowRevision())
|
||||
@@ -393,17 +393,17 @@ public final class RunVariables {
|
||||
}
|
||||
|
||||
private RunVariables(){}
|
||||
|
||||
|
||||
private record PropertyContextWithVariables(
|
||||
PropertyContext delegate,
|
||||
Map<String, Object> variables
|
||||
) implements PropertyContext {
|
||||
|
||||
|
||||
@Override
|
||||
public String render(String inline, Map<String, Object> variables) throws IllegalVariableEvaluationException {
|
||||
return delegate.render(inline, variables.isEmpty() ? this.variables : variables);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public Map<String, Object> render(Map<String, Object> inline, Map<String, Object> variables) throws IllegalVariableEvaluationException {
|
||||
return delegate.render(inline, variables.isEmpty() ? this.variables : variables);
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package io.kestra.core.runners;
|
||||
|
||||
import io.kestra.core.models.conditions.ConditionContext;
|
||||
import io.kestra.core.models.flows.Flow;
|
||||
import io.kestra.core.models.flows.FlowWithSource;
|
||||
import io.kestra.core.models.triggers.AbstractTrigger;
|
||||
import io.kestra.core.models.triggers.Trigger;
|
||||
@@ -28,7 +27,7 @@ public interface SchedulerTriggerStateInterface {
|
||||
|
||||
Trigger update(Trigger trigger);
|
||||
|
||||
Trigger update(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) throws Exception;
|
||||
Trigger update(FlowWithSource flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) throws Exception;
|
||||
|
||||
/**
|
||||
* QueueException required for Kafka implementation
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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));
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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));
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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}
|
||||
**/
|
||||
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user