POC: KEDA ScaledJob for queue jobs (payload stays in Redis)#41
Open
abnegate wants to merge 10 commits into
Open
POC: KEDA ScaledJob for queue jobs (payload stays in Redis)#41abnegate wants to merge 10 commits into
abnegate wants to merge 10 commits into
Conversation
Alternative to the env-var KubernetesJob broker: instead of inlining the payload in a per-message Job, producers enqueue to the ordinary Redis broker and a KEDA ScaledJob scales one-shot worker Jobs off the queue depth. Each worker drains the queue with the same Redis broker and exits — no custom broker, no payload on any Kubernetes object. Adds a drain worker, Dockerfile, redis + ScaledJob manifests, a kubectl-driven e2e (KedaTest) proving KEDA spawns Jobs that drain the queue, and keda-e2e.sh (kind + KEDA + redis). Verified end-to-end on kind. See servers/Keda/README.md for the comparison. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Adds a proof-of-concept for running utopia-php/queue jobs as Kubernetes Jobs via KEDA ScaledJob, keeping job payloads in the existing Redis queue (no Kubernetes involvement at enqueue time, no payload stored on K8s objects).
Changes:
- Introduces a KEDA worker image + one-shot drain worker script and a
ScaledJobmanifest for scaling off Redis list depth. - Adds an end-to-end PHPUnit test (
KedaTest) plus akind + KEDA + Redisharness script to validate the flow. - Documents the approach and trade-offs vs the env-var
KubernetesJobapproach.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/queue/tests/Queue/servers/Keda/worker.php | One-shot worker that drains Redis queue messages using the existing Redis broker. |
| packages/queue/tests/Queue/servers/Keda/README.md | Documentation for running the KEDA ScaledJob POC and trade-offs. |
| packages/queue/tests/Queue/servers/Keda/k8s.yaml | Kind-friendly namespace + Redis + KEDA ScaledJob manifest for scaling jobs from Redis depth. |
| packages/queue/tests/Queue/servers/Keda/Dockerfile | Builds the worker container image (PHP + redis extension + project code). |
| packages/queue/tests/Queue/E2E/Adapter/KedaTest.php | E2E test that enqueues directly into Redis and asserts KEDA drains the queue via Jobs. |
| packages/queue/tests/keda-e2e.sh | Harness that provisions kind + KEDA, loads the worker image, applies manifests, and runs the test. |
| packages/queue/phpunit.xml | Registers KedaTest in the e2e testsuite (skips unless KEDA_E2E=true). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Contributor
Greptile SummaryThis POC introduces a K8s-native KEDA
Confidence Score: 5/5Safe to merge as a POC; the two findings are minor asymmetries in the adapter's signal-handling lifecycle that don't affect message correctness or drain-loop behavior. The core drain loop, runsToCompletion() re-throw path, and KEDA harness are all sound. The two issues — a redundant consumer->close() in stop() that duplicates the close already done by Server's workerStop callback, and pcntl signal handlers not being cleared after drain — are lifecycle asymmetries that can produce noisy close errors or stale handlers during process teardown, but neither causes message loss or incorrect job outcomes in the normal flow. packages/queue/src/Queue/Adapter/KubernetesJob.php — the stop() double-close and missing pcntl signal cleanup Important Files Changed
Reviews (9): Last reviewed commit: "fix(queue): run the whole KubernetesJob ..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Factor the kind + KEDA + Redis + ScaledJob provisioning into tests/keda-lib.sh (keda_up/keda_down) and have both keda-e2e.sh and the package's e2e.sh use it, so `bin/monorepo test queue` (and thus CI) stands up KEDA and runs KedaTest against a real cluster instead of skipping it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- keda-lib.sh: pin + SHA-256-verify Helm (same as kind/kubectl) instead of curl|bash of an unpinned installer; only tear down a kind cluster this run created, never a pre-existing one. - KedaTest: also assert the .failed.* list is empty — draining the main queue alone doesn't prove success since receive() pops before handling; note that enqueue() mirrors Redis::enqueue()'s envelope. - k8s.yaml: correct the scaling comment (KEDA scales N Jobs, workers batch-drain) and document the backoffLimit/processing-orphan crash-recovery caveat. - README: crash-recovery section (processing-list reaper via Publisher::retry). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Member
Author
|
Addressed in 4781c8f: greptile-apps
Copilot
|
The KEDA approach still needs one bit of library code: a consumer that drains the queue and exits (so a Job completes) instead of blocking like the Swoole/ Workerman adapters. Extract that out of the test worker into Utopia\Queue\Adapter\KubernetesJob, cover it with a bare-host unit test (KubernetesJobAdapterTest), and have the KEDA worker run it via Server. Producers are unchanged — they still enqueue with any Publisher (e.g. the Redis broker). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
keda_up ran unconditionally under set -e, so a provisioning failure (no Docker, kind download failure) aborted the whole e2e suite before phpunit, skipping the Swoole/Workerman/pool tests. Gate it: required in CI (hard fail so KEDA regressions surface), best-effort locally where KedaTest self-skips. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…d termination and boot failures Register pcntl SIGTERM/SIGINT handlers in the KubernetesJob consume loop so graceful pod termination drains the in-flight message instead of being SIGKILLed (PHP is PID 1 in the Job container, so an unhandled SIGTERM is ignored). Run the workerStart callbacks under try/finally so workerStop (Timer cleanup) always runs. Add runsToCompletion() to Adapter, overridden true in KubernetesJob, and rethrow boot/consume-loop failures from Server::start() for run-to-completion adapters so the Job fails (honouring backoffLimit) rather than exiting 0. Guard the Redis broker claim writes after the pop and requeue the payload on failure so a mid-claim error no longer strands the message. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
abnegate
force-pushed
the
feat/queue-keda-poc
branch
from
d399b5c to
df8e154
Compare
php:8.4-cli-alpine ships without pcntl, so the new SIGTERM handling in Adapter\KubernetesJob threw at startup and every KEDA Job exited non-zero (the run-to-completion rethrow working as intended) before draining, failing KedaTest. Suggest ext-pcntl at the package level. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
pcntl conflicts with Swoole's signalfd, so when Swoole is loaded the drain runs inside a coroutine scheduler with a sibling coroutine on Coroutine\System::waitSignal. pcntl remains only as the non-Swoole fallback. Throwables are captured inside the scheduler and rethrown after run() since exceptions cannot cross coroutine boundaries. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Timers and signal watchers registered by workerStart hooks (e.g. a telemetry Timer::tick) must be created and cleared inside the same coroutine scheduler, or Coroutine\run never returns and the Job never completes. Coroutine::cancel on a waitSignal coroutine also leaves an orphaned signalfd watcher that swallows SIGTERM with no callback, so signals now use Process::signal registered and removed around the drain. Proven against the real cloud worker: empty-queue exit 0 in 2s, SIGTERM mid-drain graceful exit 0, message consumed off the queue. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A mid-claim Redis failure now propagates as before; the run-to-completion adapter surfaces it as a failed Job rather than silently requeueing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or 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
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.
Alternative POC to the env-var
KubernetesJobbroker (#35), built to compare the two ways of running queue jobs as Kubernetes Jobs.Approach
The K8s-native pattern: the payload stays in the Redis queue, and KEDA scales worker Jobs off the queue depth.
Utopia\Queue\Broker\Redisbroker — no custom broker, no Kubernetes involvement at enqueue time.ScaledJobwatches the queue's Redis list length and spawns one-shot worker Jobs (up tomaxReplicaCount).tests/Queue/servers/Keda/worker.php) drains the queue with the same Redis broker (receive→ handle →commit) and exits.Notably this needs zero library code — just the existing Redis broker + a drain worker + the
ScaledJobmanifest.Proven end-to-end
tests/keda-e2e.shstands up kind + KEDA (helm) + Redis + theScaledJob, loads the worker image, and runsKedaTest, which enqueues messages and asserts KEDA spawns Jobs that drain the queue. Verified on kind (OK (1 test, 4 assertions); observed 8queue-worker-*Jobs draining the queue).bin/monorepo check queuepasses (pint + phpstan + rector).env-var
KubernetesJob(#35) vs KEDAScaledJob(this PR)ARG_MAXlimits, visible in pod spec, duplicated per Podappwrite-labs/php-k8sin the appRecommendation: KEDA is the more correct/standard path (keeps the queue a queue, no payload on etcd, no custom broker); its cost is requiring the KEDA operator in the cluster. See
tests/Queue/servers/Keda/README.md.🤖 Generated with Claude Code