SLO: async-query + topic (sync/async) workloads, label-driven, with delivery/ordering validation - #851
Conversation
Async mode is now derived from the workload label (async-*); add an async query job manager and run_async path for the table runner, plus the async-query entry in the SLO CI matrix.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #851 +/- ##
=======================================
Coverage 81.29% 81.29%
=======================================
Files 94 94
Lines 12109 12109
Branches 1184 1184
=======================================
Hits 9844 9844
Misses 1802 1802
Partials 463 463
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
![robot-vibe-db[bot]](https://avatars.githubusercontent.com/in/3319454?s=60&v=4){kind=small}
There was a problem hiding this comment.
AI Review Summary
Verdict: ❌ 1 critical issue(s) found
Critical issues
- Critical | High:
async-tablelabel accepted by docker-entrypoint but crashes at runtime inAsyncTableJobManager.run_tests()—tests/slo/docker-entrypoint.sh:23
Other findings
- Major | Medium: Async metric sender pushes ~1000x too frequently vs sync (pre-existing bug propagated to new base class) —
tests/slo/src/jobs/base.py:111 - Minor | High:
asyncio.get_event_loop()deprecated since Python 3.10; should beasyncio.get_running_loop()—tests/slo/src/jobs/base.py:116 - Minor | Medium: README label table is incomplete — missing
async-table,sync-topic,async-topicthat the entrypoint accepts —tests/slo/README.md:24 - Nit | Low:
--read-threadshelp text says "write" and--write-threadssays "read" (pre-existing, not introduced by this PR) —tests/slo/src/options.py
This review was generated automatically. Critical issues require attention; other findings are advisory.
If this comment was useful, please give it a 👍 — it helps us improve the review bot.
| case "${WORKLOAD_NAME:-sync-query}" in | ||
| sync-table|sync-query) PREFIX=table ;; | ||
| topic) PREFIX=topic ;; | ||
| sync-table|sync-query|async-query|async-table) PREFIX=table ;; |
There was a problem hiding this comment.
Severity: Critical
Confidence: High
The case statement accepts async-table as a valid WORKLOAD_NAME, routing it to PREFIX=table. At runtime, _is_async_workload() detects the async- prefix and calls TableRunner.run_async(), which creates an AsyncTableJobManager. But AsyncTableJobManager.run_tests() explicitly rejects anything other than "async-query":
if self.workload_type != "async-query":
raise ValueError(f"Unsupported async workload type: {self.workload_type}")This means WORKLOAD_NAME=async-table passes entrypoint validation but crashes at runtime with a ValueError.
Either remove async-table from the case statement (if async table-service support is not yet implemented), or update AsyncTableJobManager.run_tests() to handle it.
| start_time = time.time() | ||
| logger.info("Start push metrics (async)") | ||
|
|
||
| limiter = AsyncLimiter(max_rate=10**6 // self.args.report_period, time_period=1) |
There was a problem hiding this comment.
Severity: Major
Confidence: Medium
The rate calculation 10**6 // self.args.report_period with the default report_period=1000 (milliseconds) yields max_rate=1000 pushes per second. Compare with the sync version in BaseJobManager._metric_sender() which uses SyncRateLimiter(min_interval_s=report_period_ms / 1000.0) — that gives 1 push per second at the same default.
The async metric sender pushes metrics ~1000x more frequently than the sync one. This is a pre-existing bug (moved from async_topic_jobs.py), but since this PR copies it into the shared base class where it will affect all future async workloads, it is worth fixing now.
Suggested fix:
limiter = AsyncLimiter(max_rate=1, time_period=self.args.report_period / 1000.0)| while time.time() - start_time < runtime: | ||
| async with limiter: | ||
| # Call sync metrics.push() in executor to avoid blocking the event loop. | ||
| await asyncio.get_event_loop().run_in_executor(None, self.metrics.push) |
There was a problem hiding this comment.
Severity: Minor
Confidence: High
asyncio.get_event_loop() has been deprecated since Python 3.10 and emits a DeprecationWarning in some contexts. Since this code always runs inside an async function (within a running event loop), asyncio.get_running_loop() is the correct replacement — it is guaranteed to return the running loop and never creates a new one.
Suggested fix:
await asyncio.get_running_loop().run_in_executor(None, self.metrics.push)
|
Analysis performed by claude, claude-opus-4-6. |
A labeled event for any label (e.g. the AI-review bot toggling ai_review_in_process / ai_reviewed) spawned a fresh run that cancelled the in-progress one via cancel-in-progress. Gate the job on the added label being SLO for labeled events.
slo-report fired on any non-skipped SLO workflow_run, so a run cancelled mid-flight (superseded via cancel-in-progress) still removed the SLO label and published an empty report. Gate both jobs on a genuine success/failure conclusion.
Times the local-ydb/debian pulls on the runner (invisible under the action's --quiet-build) and warms the local store. Remove before merge.
Closed-loop with read_threads workers capped throughput at workers/latency (~500 read rps at 8 workers x 16ms), never reaching the 1000 target. Switch to an open-loop generator: evenly paced submission at the target rps, each request its own task, in-flight bounded by a semaphore (backpressure). Throughput now holds the target regardless of per-request latency; read_threads/write_threads act as on/off gates.
Open-loop at MAX_INFLIGHT=256 flooded the single event loop: since the loop is CPU/throughput-bound (~480 read rps under cpus:2.0), extra concurrency just queued - read throughput fell 483->340 rps and p50 latency exploded 16->653 ms (Little's law: 256/340). The closed-loop with ~8 in-flight already sits near the single-loop ceiling at minimal latency, so restore it.
Rework the async topic workload to validate end-to-end delivery and per-producer ordering under chaos: - writers pinned to partitions (partition_id = i % N) with a stable, ref-scoped producer_id; each message carries writer_id:seqno:write_ts_ns - reader demuxes by writer_id and tracks the next expected seqno: forward gap = lost (fails via *_error* threshold), backward = duplicate (informational; reconnect redelivery), and computes end-to-end latency (same process, so the write ts is comparable) - new topic metrics (e2e latency p50/p99, delivered rps, lost, duplicates) emitted via OTLP and surfaced through tests/slo/metrics-topic.yaml (merged into the action's metrics via metrics_yaml_path) - topics are scoped per ref so current/baseline don't share a topic - async-topic added to the SLO CI matrix
Two sources of false 'lost' (first async-topic run showed avg ~81/window while delivery kept up at 100 rps and availability was 100%): - per-reader expected-seqno state saw a forward gap whenever a partition moved between readers on a consumer-group rebalance (chaos reconnect). Share the expected map across readers (safe: single event loop, no await in _validate) so a handoff continues the sequence. - a failed write_with_ack still advanced the local seqno, leaving a gap for a message that was never written. Advance seqno only on success.
Redelivered (duplicate) messages carry an old write_ts, so recording their e2e latency inflated p50 (~42ms -> ~210ms once rebalance redelivery appeared) and their reads spiked delivered_rps. Record e2e and count delivered only on the first delivery of a seqno; duplicates are counted separately.
There was a problem hiding this comment.
Pull request overview
Extends the SLO harness to support label-driven workload selection (including async-* running via ydb.aio) and hardens topic workloads for chaos testing by validating delivery/ordering and bounding all potentially blocking operations with timeouts/recreate loops.
Changes:
- Add
async-query(query service) workload implemented viaydb.aio+ async job managers. - Harden
sync-topic/async-topicworkloads with per-producer seqno validation (lost/duplicate detection) and end-to-end latency metrics. - Adjust GitHub workflows to (re)trigger SLO only when the
SLOlabel itself is added, and to publish/remove labels only on genuine success/failure (not cancelled).
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/slo/src/runners/table_runner.py | Adds async table runner entrypoint (run_async) for query-service workload. |
| tests/slo/src/root_runner.py | Derives async vs sync execution from workload label (async-*) with --async override. |
| tests/slo/src/options.py | Updates topic default partitions to 8 and retains CLI options used by new workloads. |
| tests/slo/src/jobs/topic_payload.py | Adds shared encode/decode helpers for topic payload validation. |
| tests/slo/src/jobs/topic_jobs.py | Hardens sync topic writer/reader loops, adds seqno-based validation + counters/e2e recording. |
| tests/slo/src/jobs/base.py | Introduces AsyncBaseJobManager with async metrics push task. |
| tests/slo/src/jobs/async_topic_jobs.py | Hardens async topic writer/reader loops, adds shared validation/counters/e2e recording. |
| tests/slo/src/jobs/async_table_jobs.py | Adds async query workload job manager using ydb.aio.QuerySessionPool. |
| tests/slo/src/core/metrics.py | Extends metrics with topic e2e gauges + delivered/lost/duplicated counters and recording methods. |
| tests/slo/README.md | Documents label-driven workloads and topic delivery/ordering validation semantics + new metrics. |
| tests/slo/metrics-topic.yaml | Adds PromQL queries for topic e2e/delivery/loss/duplicate metrics to merge into action defaults. |
| tests/slo/docker-entrypoint.sh | Derives per-ref topic path and supports new workload labels in container entrypoint. |
| .github/workflows/slo.yml | Adds matrix entries for async-query + sync/async topic; refines triggering to SLO label addition. |
| .github/workflows/slo-report.yml | Publishes report / removes SLO label only on success/failure (not cancelled/skipped). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| limiter = AsyncLimiter(max_rate=10**6 // self.args.report_period, time_period=1) | ||
|
|
||
| while time.time() - start_time < runtime: | ||
| async with limiter: | ||
| # Call sync metrics.push() in executor to avoid blocking the event loop. | ||
| await asyncio.get_event_loop().run_in_executor(None, self.metrics.push) |
| ts = self.metrics.start((OP_TYPE_READ,)) | ||
| try: | ||
| msg = reader.receive_message(timeout=read_timeout) | ||
| self.metrics.stop((OP_TYPE_READ,), ts) | ||
| except TimeoutError as e: | ||
| # No message within read_timeout: at steady rate this | ||
| # means the reader is starved (outage/stall), so it is | ||
| # a read failure (visible), not a silent wait. | ||
| self.metrics.stop((OP_TYPE_READ,), ts, error=e) | ||
| continue | ||
| except Exception as e: | ||
| self.metrics.stop((OP_TYPE_READ,), ts, error=e) | ||
| logger.error("Read error (recreating reader): %s", e) | ||
| break | ||
|
|
||
| if msg is None: | ||
| continue | ||
|
|
||
| self._validate(msg) | ||
|
|
||
| try: | ||
| reader.commit_with_ack(msg, timeout=read_timeout) | ||
| except Exception as e: | ||
| logger.error("Commit error: %s", e) |
| ts = self.metrics.start((OP_TYPE_READ,)) | ||
| try: | ||
| msg = await asyncio.wait_for(reader.receive_message(), read_timeout) | ||
| self.metrics.stop((OP_TYPE_READ,), ts) | ||
| except asyncio.TimeoutError as e: | ||
| # No message within read_timeout: at steady rate this | ||
| # means the reader is starved (outage/stall), so it is | ||
| # a read failure (visible), not a silent wait. | ||
| self.metrics.stop((OP_TYPE_READ,), ts, error=e) | ||
| continue | ||
| except Exception as e: | ||
| self.metrics.stop((OP_TYPE_READ,), ts, error=e) | ||
| logger.error("Read error (recreating reader): %s", e) | ||
| break | ||
|
|
||
| if msg is None: | ||
| continue | ||
|
|
||
| self._validate(msg) | ||
|
|
||
| try: | ||
| await asyncio.wait_for(reader.commit_with_ack(msg), read_timeout) | ||
| except Exception as e: | ||
| logger.error("Commit error: %s", e) |
There was a problem hiding this comment.
AI Review Summary
Verdict: ❌ 1 critical issue(s) found
Critical issues
- Major | High: Async metrics push rate is ~1000x too high due to wrong formula —
tests/slo/src/jobs/base.py:111
Other findings
- Minor | Medium: Stale messages from a cancelled previous run can corrupt the
_expectedseqno tracker, producing false duplicates and invalid e2e latency —tests/slo/src/jobs/async_topic_jobs.py:148 - Nit | Low:
asyncio.get_event_loop()is soft-deprecated since Python 3.10; preferasyncio.get_running_loop()—tests/slo/src/jobs/base.py:116
This review was generated automatically. Critical issues require attention; other findings are advisory.
If this comment was useful, please give it a 👍 — it helps us improve the review bot.
| start_time = time.time() | ||
| logger.info("Start push metrics (async)") | ||
|
|
||
| limiter = AsyncLimiter(max_rate=10**6 // self.args.report_period, time_period=1) |
There was a problem hiding this comment.
Severity: Major
Confidence: High
The rate formula 10**6 // self.args.report_period is off by 1000x. report_period is in milliseconds (default 1000, per options.py), so this computes:
10**6 // 1000 = 1000pushes per second
The sync counterpart in BaseJobManager._run_metric_job() (line 62) correctly converts:
SyncRateLimiter(min_interval_s=report_period_ms / 1000.0) # 1000ms → 1.0s → 1 push/secThe async version should mirror that rate. With the current formula, the async metrics sender pushes ~1000x more frequently than intended, which will waste CPU/network on OTLP exports and may overwhelm the Prometheus endpoint.
Suggested fix:
limiter = AsyncLimiter(max_rate=max(1, 1000 // self.args.report_period), time_period=1)Or, for non-integer rates (e.g. report_period=3000 → 0.33 pushes/sec):
limiter = AsyncLimiter(max_rate=1, time_period=max(0.001, self.args.report_period / 1000.0))|
|
||
| logger.info("Stop async topic reader %s", reader_id) | ||
|
|
||
| def _validate(self, msg) -> None: |
There was a problem hiding this comment.
Severity: Minor
Confidence: Medium
If a previous SLO run with the same ref was cancelled mid-flight (cancel-in-progress: true), its uncommitted messages may still be in the topic. When the new run's reader picks them up:
- The stale message's
seqno(e.g. 100) sets_expected[writer_id] = 101. - All subsequent real messages from the current run (seqno 1, 2, …, 100) are classified as duplicates, since
seqno < 101. - The
write_ts_nsfrom the stale message is amonotonic_ns()value from a different process invocation — the resulting e2e latency is meaningless (clamped to 1 µs or 60 s byrecord_e2e).
The same issue applies to topic_jobs.py:154 (sync variant).
Possible mitigations:
- Embed a per-run nonce (e.g. PID or UUID) in the payload header so the reader can discard messages from a previous run.
- Or call
topic-cleanup+topic-createin the entrypoint beforetopic-runto start with a clean topic.
This may not matter in practice if the consumer offset was committed past the old messages, but under cancellation the commit is not guaranteed.
| while time.time() - start_time < runtime: | ||
| async with limiter: | ||
| # Call sync metrics.push() in executor to avoid blocking the event loop. | ||
| await asyncio.get_event_loop().run_in_executor(None, self.metrics.push) |
There was a problem hiding this comment.
Severity: Nit
Confidence: Low
asyncio.get_event_loop() emits a deprecation warning in Python 3.10+ when called outside of an async context. Here it is called inside an async def, so a running loop exists and it works fine, but asyncio.get_running_loop() (available since 3.7) is the recommended replacement and communicates the intent more clearly.
await asyncio.get_running_loop().run_in_executor(None, self.metrics.push)
|
Analysis performed by claude, claude-opus-4-6. |
…old commit into read metric - docker-entrypoint: remove async-table (only async-query is implemented; it passed entrypoint validation then crashed at runtime). - base async metric sender: pushed ~1000x too often (max_rate=1e6//period); now one push per report_period (guards period=0), and use get_running_loop(). - topic readers (sync+async): the read op now spans receive+commit, so a commit failure / None message is reflected in read metrics instead of only being logged; a receive timeout still counts as a starvation error. - options: fix swapped --read/--write-threads help text.
The generic read_latency for topics is receive_message wait time (not a read cost) and is noisy, tripping the global *_latency_* regression threshold on small-number deltas; the real read-side latency is the custom topic_e2e_latency_*. Add tests/slo/thresholds-topic.yaml (read_latency_* -> direction: neutral) and wire it per topic scenario via the init thresholds_yaml_path input (ydb-slo-action#57 per-scenario thresholds). Ignored on action versions without that feature, so it is harmless now and self-activates once it lands; write_latency and table/query keep the strict global thresholds.
Under chaos the p99 is dominated by node-kill outage spikes (and async single-event-loop scheduling jitter), uncorrelated between the current and baseline containers, so the p99 current-vs-baseline delta is noise. Keep p50/p95 (write ack, e2e delivery) and delivered/lost/availability gated; only the p99 tails and the receive-wait read_latency are neutral.
🌋 SLO Test Results🔴 5 workload(s) tested — 1 workload(s) exceeded failure thresholds
Threshold violations: sync-topic:
Generated by ydb-slo-action |
Extends the SLO suite with new label-driven workloads and hardens the topic path.
Workloads (selected by the
WORKLOAD_NAMElabel;async-*runs theydb.aiopath)ydb.aio(closed-loop, mirrors the sync model for comparable metrics).producer_id; readers demux bywriter_idand track the next expected seqno (shared, so a consumer-group rebalance doesn't report a false gap). A forward gap = lost (fails the run via the*_error*threshold); a backward seqno = duplicate (reconnect redelivery, informational); plus end-to-end latency (write→read, same process). New topic metrics surfaced viatests/slo/metrics-topic.yaml(merged throughmetrics_yaml_path). Seetests/slo/TOPIC_SCENARIO.md.Hang-proofing (topic)
The SLO caught a real hang: under chaos the async topic writer wedged in the SDK's GZIP encode executor (
cannot schedule new futures after shutdown) and, awaited without a timeout, silently stalled the run at ~half duration. Hardened so this can never silently hang: everywrite_with_ack/receive_message/commit_with_ackis bounded by a timeout, the writer/reader self-heal (recreate on failure, seqno persists), and the codec isRAW(skips the buggy executor path; RAW==GZIP throughput on small messages). The underlying SDK bug is tracked in #852.Workflow fixes
SLOlabel itself is added — unrelated label changes (e.g. the AI-review bot) no longer spawn a run that cancels the in-progress one.slo-reportconsumes the label / publishes only on a genuinesuccess/failure, not oncancelled(so a superseded run no longer strips the label mid-flight).Notes
tests/slo/,ydb/(the SDK) is unchanged.