feat: Implement storage job limiter #2123
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
Matovidlo
force-pushed
the
mv-PSGO-783-implement-storage-job-rate-limiter
branch
5 times, most recently
from
69affdf to
c1ac629
Compare
Matovidlo
commented
Nov 8, 2024
internal/pkg/service/stream/storage/node/coordinator/filerotation/operator.go
Outdated
Show resolved
Hide resolved
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/test/table.go
Show resolved
Hide resolved
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/file.go
Outdated
Show resolved
Hide resolved
Matovidlo
requested review from
jachym-tousek-keboola and
hosekpeter
as code owners
hosekpeter
reviewed
Nov 11, 2024
internal/pkg/service/stream/storage/model/repository/job/job_list.go
Outdated
Show resolved
Hide resolved
internal/pkg/service/stream/storage/level/target/config/config.go
Outdated
Show resolved
Hide resolved
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/job.go
Outdated
Show resolved
Hide resolved
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/job.go
Outdated
Show resolved
Hide resolved
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/test/table.go
Show resolved
Hide resolved
Matovidlo
commented
Nov 11, 2024
Used in job repository, when deleting sink, delete all jobs under the sinkKey
The job is deleted once success status is returned from GET jobs connection API endpoint
When exceeding jobs limit, do not rotate file, instead return and accumulate slices further. When storage job is deleted by metadata cleanup module, the import is executed
When not present, the test never ends due to set limit
Matovidlo
force-pushed
the
mv-PSGO-783-implement-storage-job-rate-limiter
branch
from
60557bb to
1b280ed
Compare
internal/pkg/service/stream/storage/node/coordinator/filerotation/conditions.go
Outdated
Show resolved
Hide resolved
internal/pkg/service/stream/storage/node/coordinator/filerotation/conditions.go
Outdated
Show resolved
Hide resolved
Create mirror on top of bridge component. Add new throttle plugin and evaluate it in filerotation module. Adjust tests
call it in coordinator. Fix fileroration operator to evaluate sink throttling after import conditions are satisfied
jachym-tousek-keboola
requested changes
Nov 18, 2024
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/model/job.go
Outdated
Show resolved
Hide resolved
...nal/pkg/service/stream/sink/type/tablesink/keboola/bridge/model/repository/job/job_create.go
Outdated
Show resolved
Hide resolved
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/test/table.go
Show resolved
Hide resolved
internal/pkg/service/stream/storage/node/coordinator/coordinator.go
Outdated
Show resolved
Hide resolved
Mirror uses context instead of context.Background()
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/bridge.go
Show resolved
Hide resolved
Matovidlo
force-pushed
the
mv-PSGO-783-implement-storage-job-rate-limiter
branch
from
0332f7e to
c78fba5
Compare
Rework metadata cleanup test to store token so it can test the workflow of metadata cleanup correctly.
Matovidlo
force-pushed
the
mv-PSGO-783-implement-storage-job-rate-limiter
branch
from
c78fba5 to
41c0053
Compare
jachym-tousek-keboola
requested changes
Nov 19, 2024
internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/job.go
Outdated
Show resolved
Hide resolved
Use it from metadatacleanup module
Stream Kubernetes Diff [CI]Between Expand--- /tmp/artifacts/test-k8s-state.old.json.processed.kv 2024-11-20 10:33:47.730683078 +0000
+++ /tmp/artifacts/test-k8s-state.new.json.processed.kv 2024-11-20 10:33:48.210681617 +0000
@@ -200 +200 @@
-<Deployment/stream-api>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Deployment/stream-api>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -364 +364 @@
-<Deployment/stream-http-source>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Deployment/stream-http-source>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -525 +525 @@
-<Deployment/stream-storage-coordinator>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Deployment/stream-storage-coordinator>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -602 +602 @@
-<Endpoints/stream-etcd-headless>.subsets[0].addresses[0].hostname = "stream-etcd-1";
+<Endpoints/stream-etcd-headless>.subsets[0].addresses[0].hostname = "stream-etcd-2";
@@ -606 +606 @@
-<Endpoints/stream-etcd-headless>.subsets[0].addresses[0].targetRef.name = "stream-etcd-1";
+<Endpoints/stream-etcd-headless>.subsets[0].addresses[0].targetRef.name = "stream-etcd-2";
@@ -609 +609 @@
-<Endpoints/stream-etcd-headless>.subsets[0].addresses[1].hostname = "stream-etcd-2";
+<Endpoints/stream-etcd-headless>.subsets[0].addresses[1].hostname = "stream-etcd-1";
@@ -613 +613 @@
-<Endpoints/stream-etcd-headless>.subsets[0].addresses[1].targetRef.name = "stream-etcd-2";
+<Endpoints/stream-etcd-headless>.subsets[0].addresses[1].targetRef.name = "stream-etcd-1";
@@ -653 +653 @@
-<Endpoints/stream-etcd>.subsets[0].addresses[0].targetRef.name = "stream-etcd-1";
+<Endpoints/stream-etcd>.subsets[0].addresses[0].targetRef.name = "stream-etcd-2";
@@ -659 +659 @@
-<Endpoints/stream-etcd>.subsets[0].addresses[1].targetRef.name = "stream-etcd-2";
+<Endpoints/stream-etcd>.subsets[0].addresses[1].targetRef.name = "stream-etcd-1";
@@ -717 +717 @@
-<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[0].hostname = "stream-storage-writer-reader-0";
+<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[0].hostname = "stream-storage-writer-reader-1";
@@ -721 +721 @@
-<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[0].targetRef.name = "stream-storage-writer-reader-0";
+<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[0].targetRef.name = "stream-storage-writer-reader-1";
@@ -724 +724 @@
-<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[1].hostname = "stream-storage-writer-reader-1";
+<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[1].hostname = "stream-storage-writer-reader-0";
@@ -728 +728 @@
-<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[1].targetRef.name = "stream-storage-writer-reader-1";
+<Endpoints/stream-storage-writer-reader>.subsets[0].addresses[1].targetRef.name = "stream-storage-writer-reader-0";
@@ -1214,2 +1214,2 @@
-<Pod/stream-api-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
-<Pod/stream-api-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Pod/stream-api-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
+<Pod/stream-api-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -1534 +1534 @@
-<Pod/stream-etcd-0>.spec.containers[0].env[21].value = "new";
+<Pod/stream-etcd-0>.spec.containers[0].env[21].value = "existing";
@@ -1780 +1780 @@
-<Pod/stream-etcd-1>.spec.containers[0].env[21].value = "new";
+<Pod/stream-etcd-1>.spec.containers[0].env[21].value = "existing";
@@ -2026 +2026 @@
-<Pod/stream-etcd-2>.spec.containers[0].env[21].value = "new";
+<Pod/stream-etcd-2>.spec.containers[0].env[21].value = "existing";
@@ -2350,2 +2350,2 @@
-<Pod/stream-http-source-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
-<Pod/stream-http-source-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Pod/stream-http-source-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
+<Pod/stream-http-source-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -2742,2 +2742,2 @@
-<Pod/stream-storage-coordinator-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
-<Pod/stream-storage-coordinator-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Pod/stream-storage-coordinator-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
+<Pod/stream-storage-coordinator-<hash>>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -2978 +2978 @@
-<Pod/stream-storage-writer-reader-0>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Pod/stream-storage-writer-reader-0>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -3061 +3061 @@
-<Pod/stream-storage-writer-reader-0>.spec.containers[1].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Pod/stream-storage-writer-reader-0>.spec.containers[1].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -3231 +3231 @@
-<Pod/stream-storage-writer-reader-1>.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Pod/stream-storage-writer-reader-1>.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -3314 +3314 @@
-<Pod/stream-storage-writer-reader-1>.spec.containers[1].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<Pod/stream-storage-writer-reader-1>.spec.containers[1].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -3552 +3552 @@
-<ReplicaSet/stream-api-<hash>>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<ReplicaSet/stream-api-<hash>>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -3723 +3723 @@
-<ReplicaSet/stream-http-source-<hash>>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<ReplicaSet/stream-http-source-<hash>>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -3891 +3891 @@
-<ReplicaSet/stream-storage-coordinator-<hash>>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<ReplicaSet/stream-storage-coordinator-<hash>>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -3932,0 +3933,12 @@
+<Secret/sh.helm.release.v1.stream-etcd.v2> = {};
+<Secret/sh.helm.release.v1.stream-etcd.v2>.apiVersion = "v1";
+<Secret/sh.helm.release.v1.stream-etcd.v2>.data = {};
+<Secret/sh.helm.release.v1.stream-etcd.v2>.kind = "Secret";
+<Secret/sh.helm.release.v1.stream-etcd.v2>.metadata = {};
+<Secret/sh.helm.release.v1.stream-etcd.v2>.metadata.labels = {};
+<Secret/sh.helm.release.v1.stream-etcd.v2>.metadata.labels.name = "stream-etcd";
+<Secret/sh.helm.release.v1.stream-etcd.v2>.metadata.labels.owner = "helm";
+<Secret/sh.helm.release.v1.stream-etcd.v2>.metadata.labels.version = "2";
+<Secret/sh.helm.release.v1.stream-etcd.v2>.metadata.name = "sh.helm.release.v1.stream-etcd.v2";
+<Secret/sh.helm.release.v1.stream-etcd.v2>.metadata.namespace = "stream";
+<Secret/sh.helm.release.v1.stream-etcd.v2>.type = "helm.sh/release.v1";
@@ -4228 +4240 @@
-<StatefulSet/stream-etcd>.spec.template.spec.containers[0].env[21].value = "new";
+<StatefulSet/stream-etcd>.spec.template.spec.containers[0].env[21].value = "existing";
@@ -4478 +4490 @@
-<StatefulSet/stream-storage-writer-reader>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<StatefulSet/stream-storage-writer-reader>.spec.template.spec.containers[0].image = "docker.io/keboola/stream-api:665e97a-1732098303";
@@ -4558 +4570 @@
-<StatefulSet/stream-storage-writer-reader>.spec.template.spec.containers[1].image = "docker.io/keboola/stream-api:e375dc2-1732097778";
+<StatefulSet/stream-storage-writer-reader>.spec.template.spec.containers[1].image = "docker.io/keboola/stream-api:665e97a-1732098303";
(see artifacts in the Github Action for more information) |
jachym-tousek-keboola
approved these changes
Nov 20, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Jira: PSGO-783
Changes:
filerotationcalledsinkThrottled.sinkLimitinherits fromjobLimit, but it is configurable forfilerotationtests.OnSinkDeletewhich is connected withstoragejob repository module.storagejob repository. Whensinkis deleted, remove all jobs under thatsinkKey.import-asyncPOST endpoint interaction.filerotationmodule, which is independent ofkeboolasink/bridgemodule.sinkLimitwhen created undersinkKey. (mirror could be removed)keboolasinkandmodelmodules.keboola_test.goE2E test. We need it to unblock the sink imports. Also we want to test that the storage job limiter works correctly.Questions:
activeprefix. It can be omitted but don't, know whether we would like to inspect it ?sinkLimitalso underfileentity. Probablly will be used by next PR when we would like to propagate error into UI.