diff --git a/.github/workflows/slo.yml b/.github/workflows/slo.yml index 8c066a3d..67231712 100644 --- a/.github/workflows/slo.yml +++ b/.github/workflows/slo.yml @@ -41,6 +41,14 @@ jobs: command: "--write-rps 200 --write-threads 8 --read-threads 8" metrics_yaml_path: sdk-current/tests/slo/metrics-topic.yaml thresholds_yaml_path: sdk-current/tests/slo/thresholds-topic.yaml + - name: sync-topic-tx + command: "--write-rps 50 --write-threads 8 --read-threads 8 --messages-per-tx 10" + metrics_yaml_path: sdk-current/tests/slo/metrics-topic.yaml + thresholds_yaml_path: sdk-current/tests/slo/thresholds-topic.yaml + - name: async-topic-tx + command: "--write-rps 50 --write-threads 8 --read-threads 8 --messages-per-tx 10" + metrics_yaml_path: sdk-current/tests/slo/metrics-topic.yaml + thresholds_yaml_path: sdk-current/tests/slo/thresholds-topic.yaml concurrency: group: slo-${{ github.ref }}-${{ matrix.sdk.name }} diff --git a/tests/slo/README.md b/tests/slo/README.md index 88d46e72..5aa1e6cd 100644 --- a/tests/slo/README.md +++ b/tests/slo/README.md @@ -17,13 +17,15 @@ The workload is selected by a single label (`WORKLOAD_NAME` env var, also the execution mode is derived from the label itself — an `async-*` label runs the async (`ydb.aio`) path: -| Label | Service | Mode | -|----------------|---------------|-------| -| `sync-table` | Table service | sync | -| `sync-query` | Query service | sync | -| `async-query` | Query service | async | -| `sync-topic` | Topic service | sync | -| `async-topic` | Topic service | async | +| Label | Service | Mode | +|------------------|-----------------------|-------| +| `sync-table` | Table service | sync | +| `sync-query` | Query service | sync | +| `async-query` | Query service | async | +| `sync-topic` | Topic service | sync | +| `async-topic` | Topic service | async | +| `sync-topic-tx` | Topic + Query (tx) | sync | +| `async-topic-tx` | Topic + Query (tx) | async | > The `--async` CLI flag is kept as a manual override for `*-run` commands. > The bare `topic` label is still accepted as an alias for `sync-topic`. @@ -269,6 +271,25 @@ When running `topic-run` (`sync-topic` / `async-topic`), the program creates `re Each message carries `writer_id:seqno:write_ts_ns:` followed by padding to the configured size. Topics are scoped per ref so the current and baseline containers (same cluster, run in parallel) don't share a topic. +### Transactional topic ↔ table workload + +When running `topic-run` under `sync-topic-tx` / `async-topic-tx`, both ends of the +pipeline run inside a YDB transaction, validating **exactly-once** delivery of a +topic into a table under chaos: + +- `writeJob` — a `tx_writer` writes a batch of messages to the topic inside a + transaction (a transactional topic write); `seqno` advances only on a successful + commit (a chaos abort leaves no gap). +- `readJob` — `receive_batch_with_tx` reads a batch while the same transaction + UPSERTs each message into a sink table keyed by `(writer_id, seqno)` (topic → + table). The commit advances the topic offset and writes the sink rows atomically, + so a rolled-back tx re-reads and re-UPSERTs the same keys — idempotent, + exactly-once. + +The run must finish with `topic_lost_errors == 0`. The topic and the sink table are +ref-scoped. See `TOPIC_TX_SCENARIO.md` for the full design; extra options: +`--messages-per-tx`, `--sink-table`. + ## Collected metrics - `oks` - amount of OK requests - `not_oks` - amount of not OK requests diff --git a/tests/slo/TOPIC_TX_SCENARIO.md b/tests/slo/TOPIC_TX_SCENARIO.md new file mode 100644 index 00000000..43ba9e67 --- /dev/null +++ b/tests/slo/TOPIC_TX_SCENARIO.md @@ -0,0 +1,101 @@ +# Topic-tx SLO scenario — transactional topic ↔ table pipeline + +What the `sync-topic-tx` / `async-topic-tx` workloads do: an **exactly-once** +processing pipeline where both ends run inside a YDB transaction. Producers write +to the topic with a `tx_writer` (a transactional topic write); consumers read from +the topic with `receive_batch_with_tx` and, in the *same* transaction, UPSERT the +message into a sink table. The commit atomically advances the topic offset and +persists the sink rows — so every produced message lands in the sink table exactly +once, even while a chaos monkey keeps killing YDB nodes. + +## Topology + +``` + producer tx (transactional topic write) per-ref topic: /Root/testdb/slo_topic_tx_ + │ tx_writer(tx).write(msg); commit ┌──────────────────────────────────────────────┐ + w0 ──┼──produce p0──▶ partition 0 ────────────────▶│ p0: m1 m2 m3 … (server-ordered per producer) │ + w1 ──┴──produce p1──▶ partition 1 ────────────────▶│ p1: m1 m2 m3 … │ + └───────────────────────┬──────────────────────┘ + │ one consumer group + ┌──────────────────────────────────────────▼──────────────────────┐ + │ consumer tx: receive_batch_with_tx(tx) → decode → validate │ + │ → UPSERT sink(writer_id, seqno); commit │ + └───────────────────────────────────────────┬───────────────────────┘ + ▲ ▼ sink(writer_id, seqno) — exactly once + chaos-monkey: kills a YDB node every ~20-60s +``` + +Each message is self-describing: `writer_id : seqno : write_ts_ns : ` +(same codec as the plain topic scenario, `jobs/topic_payload.py`). + +## Flow (happy path + chaos) + +```mermaid +sequenceDiagram + autonumber + participant W as Producer tx (tx_writer) + participant S as YDB topic · partition i + participant R as Consumer tx (receive_batch_with_tx) + participant T as sink table + participant M as Metrics (OTLP) + + Note over W: retry_tx: tx_writer.write(msgs), commit + loop until --time + W->>S: tx_writer.write(payload) (buffered, flushed on commit) + alt commit ok + S-->>W: topic write committed + Note right of W: seqno advances only on commit (no gap) + else Unavailable (chaos) + Note right of W: retry_tx re-runs callee; nothing persisted (no gap) + end + end + + loop drain as fast as messages arrive + S-->>R: receive_batch_with_tx(tx) + R->>T: UPSERT sink(writer_id, seqno) — keyed, idempotent + alt commit ok + R->>M: for each msg: delivered++, e2e = now - write_ts, validate seqno + Note right of R: offset advance + sink rows committed atomically + else Unavailable (chaos) + R->>M: retry_tx re-reads from last committed offset + Note right of R: rolled-back tx advanced no offset ⇒ re-processed, upsert idempotent + end + end +``` + +## Invariants we assert + +| Signal | Meaning | Expectation | +|--------|---------|-------------| +| `topic_lost_errors` | forward gap in a producer's seqno (real loss) | **0** — hard-fails the run, under chaos | +| exactly-once into sink | sink keyed by `(writer_id, seqno)` — idempotent UPSERT | re-processing after a rollback creates no duplicate row | +| `topic_e2e_latency_p50/p99` | write-commit → read-commit latency | reported | +| `topic_duplicates` | redelivery after a rolled-back tx | informational (at-least-once at the app level) | +| `read/write_availability` | liveness under chaos | ~100% | + +## Why it stays correct under chaos + +- **The topic read and the table write share one transaction.** A consumer commit + advances the topic offset *and* writes the sink row atomically; an abort (chaos) + rolls back both. There is no window where one happened without the other. +- **seqno advances only after a successful commit** — an aborted/retried producer + tx leaves no gap. Validation reads the seqno embedded in the payload, so a + retried commit is at worst an app-level duplicate, never a false loss. +- **sink is keyed by `(writer_id, seqno)`** — a consumer tx that rolls back + (chaos) re-reads the same messages from the last committed offset and + re-UPSERTs the same keys: idempotent, so the table stays exactly-once. +- **validation runs only after the commit returns** — only durable offsets/rows + feed the per-producer next-expected-seqno check (shared across readers, since a + partition can move between them on rebalance). +- **per-ref topic + per-ref sink table** — the `current` and `baseline` containers + run on the same cluster in parallel; ref-scoping keeps their validation from + mixing. + +## Not covered here + +- Reading the sink table back for an independent full-table exactly-once audit + (validation is the in-memory per-producer next-expected-seqno check, fed only by + committed batches; the atomic offset+row commit makes that a faithful proxy). +- Snapshot/read-only tx modes — the pipeline runs serializable read-write. +- Deliberate non-retriable failures (SchemeError/BadRequest) — those are code/config + bugs, out of scope for a liveness SLO. diff --git a/tests/slo/docker-entrypoint.sh b/tests/slo/docker-entrypoint.sh index 55d933ec..cff86205 100755 --- a/tests/slo/docker-entrypoint.sh +++ b/tests/slo/docker-entrypoint.sh @@ -6,6 +6,7 @@ # # Inputs come from the env vars injected by the action: # WORKLOAD_NAME sync-table | sync-query | async-query | sync-topic | async-topic +# | sync-topic-tx | async-topic-tx # WORKLOAD_DURATION run duration in seconds # YDB_ENDPOINT grpc://ydb:2136 # YDB_DATABASE /Root/testdb @@ -21,7 +22,7 @@ set -e case "${WORKLOAD_NAME:-sync-query}" in sync-table|sync-query|async-query) PREFIX=table ;; - topic|sync-topic|async-topic) PREFIX=topic ;; + topic|sync-topic|async-topic|sync-topic-tx|async-topic-tx) PREFIX=topic ;; *) echo "Unknown WORKLOAD_NAME: ${WORKLOAD_NAME}" >&2 exit 1 @@ -41,7 +42,17 @@ EXTRA_ARGS="" if [ "$PREFIX" = "topic" ]; then REF_RAW="${WORKLOAD_REF:-${REF:-main}}" SAFE_REF=$(printf '%s' "$REF_RAW" | tr -c 'a-zA-Z0-9_' '_') - EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_${SAFE_REF}" + case "${WORKLOAD_NAME:-}" in + *topic-tx) + # The transactional pipeline gets its own topic plus a ref-scoped + # sink table, so current/baseline don't collide. + EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_tx_${SAFE_REF}" + EXTRA_ARGS="${EXTRA_ARGS} --sink-table slo_topic_tx_sink_${SAFE_REF}" + ;; + *) + EXTRA_ARGS="--path ${DATABASE%/}/slo_topic_${SAFE_REF}" + ;; + esac fi # Schema prep is idempotent at the SDK level for topics; for tables, a parallel diff --git a/tests/slo/src/core/metrics.py b/tests/slo/src/core/metrics.py index 313e1683..58da8ff9 100644 --- a/tests/slo/src/core/metrics.py +++ b/tests/slo/src/core/metrics.py @@ -68,6 +68,13 @@ def measure(self, labels): finally: self.stop(labels, start_ts, error=error) + def cancel(self, labels, start_time: float) -> None: + """Undo a ``start()`` for an operation that turned out to be a no-op + (e.g. a transactional reader that had no batch to process): balances the + pending counter without recording a success or a failure, so idle waits + do not count against availability.""" + return None + # --- Topic workload extensions (no-ops unless overridden) --- def record_e2e(self, seconds: float) -> None: """Record an end-to-end (write -> read) message latency, in seconds.""" @@ -252,6 +259,10 @@ def stop( with self._lock: self._get_hdr(op_type, op_status).record_value(duration_us) + def cancel(self, labels, start_time: float) -> None: + labels_t = _normalize_labels(labels) + self._pending.add(-1, attributes={"ref": REF, "operation_type": labels_t[0]}) + def push(self) -> None: with self._lock: for (op_type, op_status), hist in self._hdr.items(): diff --git a/tests/slo/src/jobs/async_topic_jobs.py b/tests/slo/src/jobs/async_topic_jobs.py index c4cfe691..e39df011 100644 --- a/tests/slo/src/jobs/async_topic_jobs.py +++ b/tests/slo/src/jobs/async_topic_jobs.py @@ -150,7 +150,9 @@ async def _run_topic_reads(self, reader_id: int): logger.info("Stop async topic reader %s", reader_id) def _validate(self, msg) -> None: - decoded = decode_payload(msg.data) + self._record(decode_payload(msg.data)) + + def _record(self, decoded) -> None: if decoded is None: self.metrics.inc_delivered() return diff --git a/tests/slo/src/jobs/async_topic_tx_jobs.py b/tests/slo/src/jobs/async_topic_tx_jobs.py new file mode 100644 index 00000000..b127e096 --- /dev/null +++ b/tests/slo/src/jobs/async_topic_tx_jobs.py @@ -0,0 +1,155 @@ +import asyncio +import logging +import time + +from core.metrics import OP_TYPE_READ, OP_TYPE_WRITE, REF + +import ydb +import ydb.aio + +from .async_topic_jobs import AsyncTopicJobManager +from .topic_payload import decode_payload, encode_payload +from .topic_tx_jobs import SINK_UPSERT_TEMPLATE + +logger = logging.getLogger(__name__) + + +class AsyncTopicTxJobManager(AsyncTopicJobManager): + """Async mirror of :class:`TopicTxJobManager` (transactional topic <-> table + pipeline). See that class for the scenario description.""" + + def __init__(self, driver, args, metrics, sink_table): + super().__init__(driver, args, metrics) + self.sink_table = sink_table + self.messages_per_tx = max(1, int(getattr(self.args, "messages_per_tx", 10))) + self._sink_upsert_query = SINK_UPSERT_TEMPLATE.format(sink_table) + self.retry_settings = ydb.RetrySettings(idempotent=True) + + # --- shared query helper -------------------------------------------------- + + async def _upsert_sink(self, tx, decoded): + writer_id, seqno, write_ts_ns = decoded + async with await tx.execute( + self._sink_upsert_query, + { + "$writer_id": (writer_id, ydb.PrimitiveType.Uint64), + "$seqno": (seqno, ydb.PrimitiveType.Uint64), + "$write_ts_ns": (write_ts_ns, ydb.PrimitiveType.Uint64), + }, + commit_tx=False, + ) as result_sets: + async for _ in result_sets: + pass + + # --- producer (transactional topic write) -------------------------------- + + async def _run_topic_writes(self, writer_id, limiter): + start_time = time.time() + partition_id = writer_id % self.partitions + producer_id = f"{REF}-p{writer_id}" + logger.info("Start async topic tx writer %s (partition %s, producer %s)", writer_id, partition_id, producer_id) + + # `seqno` lives across recreations and advances only after a successful + # commit, so the sequence the reader sees is gap-free. + seqno = 1 + while time.time() - start_time < self.args.time: + try: + async with ydb.aio.QuerySessionPool(self.aio_driver) as pool: + while time.time() - start_time < self.args.time: + base_seqno = seqno + payloads = [] + for k in range(self.messages_per_tx): + if time.time() - start_time >= self.args.time: + break + async with limiter: + payloads.append( + encode_payload( + writer_id, base_seqno + k, time.monotonic_ns(), self.args.message_size + ) + ) + if not payloads: + break + + async def callee(tx): + tx_writer = self.aio_driver.topic_client.tx_writer( + tx, + self.args.path, + producer_id=producer_id, + partition_id=partition_id, + codec=ydb.TopicCodec.RAW, + ) + for payload in payloads: + await tx_writer.write(ydb.TopicWriterMessage(data=payload)) + + ts = self.metrics.start((OP_TYPE_WRITE,)) + try: + await pool.retry_tx_async(callee, retry_settings=self.retry_settings) + self.metrics.stop((OP_TYPE_WRITE,), ts) + seqno = base_seqno + len(payloads) # advance only on commit + except Exception as e: + self.metrics.stop((OP_TYPE_WRITE,), ts, error=e) + logger.error("Tx write error (recreating writer): %s", e) + break + except Exception as e: + logger.error("Topic tx writer %s recreate: %s", writer_id, e) + await asyncio.sleep(0.2) + + logger.info("Stop async topic tx writer %s", writer_id) + + # --- consumer (topic -> table) ------------------------------------------- + + async def _run_topic_reads(self, reader_id): + start_time = time.time() + read_timeout = self.args.read_timeout / 1000 + logger.info("Start async topic tx reader %s", reader_id) + + while time.time() - start_time < self.args.time: + try: + async with ydb.aio.QuerySessionPool(self.aio_driver) as pool: + async with self.aio_driver.topic_client.reader(self.args.path, self.args.consumer) as reader: + while time.time() - start_time < self.args.time: + committed = [] + + async def callee(tx): + nonlocal committed + committed = [] + batch = await asyncio.wait_for( + reader.receive_batch_with_tx(tx, max_messages=self.messages_per_tx), read_timeout + ) + if batch is None: + return + for msg in batch.messages: + decoded = decode_payload(msg.data) + if decoded is not None: + await self._upsert_sink(tx, decoded) + committed.append(decoded) + + ts = self.metrics.start((OP_TYPE_READ,)) + try: + await pool.retry_tx_async(callee, retry_settings=self.retry_settings) + except asyncio.TimeoutError: + # No batch within read_timeout: an idle reader or + # the SDK transparently reconnecting under chaos, + # not a read failure. Don't count the op (real + # errors surface as exceptions the tx retries). + self.metrics.cancel((OP_TYPE_READ,), ts) + continue + except Exception as e: + self.metrics.stop((OP_TYPE_READ,), ts, error=e) + logger.error("Tx read error (recreating reader): %s", e) + break + + if not committed: + self.metrics.cancel((OP_TYPE_READ,), ts) + continue + + # A batch committed: the offsets/rows are durable, so + # the seqnos are exactly-once delivered. + self.metrics.stop((OP_TYPE_READ,), ts) + for decoded in committed: + self._record(decoded) + except Exception as e: + logger.error("Topic tx reader %s recreate: %s", reader_id, e) + await asyncio.sleep(0.2) + + logger.info("Stop async topic tx reader %s", reader_id) diff --git a/tests/slo/src/jobs/topic_jobs.py b/tests/slo/src/jobs/topic_jobs.py index 21645857..325ef164 100644 --- a/tests/slo/src/jobs/topic_jobs.py +++ b/tests/slo/src/jobs/topic_jobs.py @@ -156,7 +156,9 @@ def _run_topic_reads(self, reader_id): logger.info("Stop topic reader %s", reader_id) def _validate(self, msg): - decoded = decode_payload(msg.data) + self._record(decode_payload(msg.data)) + + def _record(self, decoded): if decoded is None: self.metrics.inc_delivered() return diff --git a/tests/slo/src/jobs/topic_tx_jobs.py b/tests/slo/src/jobs/topic_tx_jobs.py new file mode 100644 index 00000000..06313fcb --- /dev/null +++ b/tests/slo/src/jobs/topic_tx_jobs.py @@ -0,0 +1,169 @@ +import logging +import time + +from core.metrics import OP_TYPE_READ, OP_TYPE_WRITE, REF + +import ydb + +from .topic_jobs import TopicJobManager +from .topic_payload import decode_payload, encode_payload + +logger = logging.getLogger(__name__) + +SINK_UPSERT_TEMPLATE = """ +DECLARE $writer_id AS Uint64; +DECLARE $seqno AS Uint64; +DECLARE $write_ts_ns AS Uint64; +UPSERT INTO `{}` (writer_id, seqno, write_ts_ns, received_at) +VALUES ($writer_id, $seqno, $write_ts_ns, CurrentUtcTimestamp()); +""" + + +class TopicTxJobManager(TopicJobManager): + """Transactional topic <-> table pipeline (sync). + + Producer: a ``tx_writer`` writes a batch of messages to the topic inside a + transaction (a transactional topic write). Consumer: + ``receive_batch_with_tx`` reads a batch while the same transaction UPSERTs + each message into a sink table keyed by ``(writer_id, seqno)``. The commit + advances the topic offset and persists the sink rows atomically -> a + rolled-back tx re-reads and re-UPSERTs the same keys (idempotent), so the + table receives every message exactly once, even under chaos. + """ + + def __init__(self, driver, args, metrics, sink_table): + super().__init__(driver, args, metrics) + self.sink_table = sink_table + self.messages_per_tx = max(1, int(getattr(self.args, "messages_per_tx", 10))) + self._sink_upsert_query = SINK_UPSERT_TEMPLATE.format(sink_table) + self.retry_settings = ydb.RetrySettings(idempotent=True) + + # --- shared query helper -------------------------------------------------- + + def _upsert_sink(self, tx, decoded): + writer_id, seqno, write_ts_ns = decoded + with tx.execute( + self._sink_upsert_query, + { + "$writer_id": (writer_id, ydb.PrimitiveType.Uint64), + "$seqno": (seqno, ydb.PrimitiveType.Uint64), + "$write_ts_ns": (write_ts_ns, ydb.PrimitiveType.Uint64), + }, + commit_tx=False, + ) as result_sets: + for _ in result_sets: + pass + + # --- producer (transactional topic write) -------------------------------- + + def _run_topic_writes(self, writer_id, limiter): + start_time = time.time() + partition_id = writer_id % self.partitions + producer_id = f"{REF}-p{writer_id}" + logger.info("Start topic tx writer %s (partition %s, producer %s)", writer_id, partition_id, producer_id) + + # `seqno` lives across recreations and advances only after a successful + # commit, so the sequence the reader sees is gap-free (a failed/aborted + # tx leaves no gap; a retried commit is at worst an app-level duplicate). + seqno = 1 + while time.time() - start_time < self.args.time: + try: + with ydb.QuerySessionPool(self.driver) as pool: + while time.time() - start_time < self.args.time: + base_seqno = seqno + payloads = [] + for k in range(self.messages_per_tx): + if time.time() - start_time >= self.args.time: + break + with limiter: + payloads.append( + encode_payload( + writer_id, base_seqno + k, time.monotonic_ns(), self.args.message_size + ) + ) + if not payloads: + break + + def callee(tx): + tx_writer = self.driver.topic_client.tx_writer( + tx, + self.args.path, + producer_id=producer_id, + partition_id=partition_id, + codec=ydb.TopicCodec.RAW, + ) + for payload in payloads: + tx_writer.write(ydb.TopicWriterMessage(data=payload)) + + ts = self.metrics.start((OP_TYPE_WRITE,)) + try: + pool.retry_tx_sync(callee, retry_settings=self.retry_settings) + self.metrics.stop((OP_TYPE_WRITE,), ts) + seqno = base_seqno + len(payloads) # advance only on commit + except Exception as e: + self.metrics.stop((OP_TYPE_WRITE,), ts, error=e) + logger.error("Tx write error (recreating writer): %s", e) + break + except Exception as e: + logger.error("Topic tx writer %s recreate: %s", writer_id, e) + time.sleep(0.2) + + logger.info("Stop topic tx writer %s", writer_id) + + # --- consumer (topic -> table) ------------------------------------------- + + def _run_topic_reads(self, reader_id): + start_time = time.time() + read_timeout = self.args.read_timeout / 1000 + logger.info("Start topic tx reader %s", reader_id) + + while time.time() - start_time < self.args.time: + try: + with ydb.QuerySessionPool(self.driver) as pool: + with self.driver.topic_client.reader(self.args.path, self.args.consumer) as reader: + while time.time() - start_time < self.args.time: + committed = [] + + def callee(tx): + nonlocal committed + committed = [] + batch = reader.receive_batch_with_tx( + tx, max_messages=self.messages_per_tx, timeout=read_timeout + ) + if batch is None: + return + for msg in batch.messages: + decoded = decode_payload(msg.data) + if decoded is not None: + self._upsert_sink(tx, decoded) + committed.append(decoded) + + ts = self.metrics.start((OP_TYPE_READ,)) + try: + pool.retry_tx_sync(callee, retry_settings=self.retry_settings) + except TimeoutError: + # No batch within read_timeout: an idle reader or + # the SDK transparently reconnecting under chaos, + # not a read failure. Don't count the op (real + # errors surface as exceptions the tx retries). + self.metrics.cancel((OP_TYPE_READ,), ts) + continue + except Exception as e: + self.metrics.stop((OP_TYPE_READ,), ts, error=e) + logger.error("Tx read error (recreating reader): %s", e) + break + + if not committed: + self.metrics.cancel((OP_TYPE_READ,), ts) + continue + + # A batch committed: the offsets/rows are durable, so + # the seqnos are exactly-once delivered. + self.metrics.stop((OP_TYPE_READ,), ts) + for decoded in committed: + self._record(decoded) + except Exception as e: + logger.error("Topic tx reader %s recreate: %s", reader_id, e) + time.sleep(0.2) + + logger.info("Stop topic tx reader %s", reader_id) diff --git a/tests/slo/src/options.py b/tests/slo/src/options.py index 200c3c55..7cd5e034 100644 --- a/tests/slo/src/options.py +++ b/tests/slo/src/options.py @@ -97,6 +97,16 @@ def make_table_cleanup_parser(subparsers): add_common_options(table_cleanup_parser) +def add_topic_tx_options(parser): + """Options for the transactional topic <-> table workloads (``*-topic-tx``).""" + parser.add_argument( + "--sink-table", + default="slo_topic_tx_sink", + type=str, + help="Sink table for the transactional consumer (exactly-once target)", + ) + + def make_topic_create_parser(subparsers): topic_create_parser = subparsers.add_parser("topic-create", help="Create topic with consumer") add_common_options(topic_create_parser) @@ -104,6 +114,7 @@ def make_topic_create_parser(subparsers): topic_create_parser.add_argument("--path", default="/local/slo_topic", type=str, help="Topic path") topic_create_parser.add_argument("--consumer", default="slo_consumer", type=str, help="Topic consumer name") topic_create_parser.add_argument("--partitions-count", default=8, type=int, help="Partition count") + add_topic_tx_options(topic_create_parser) def make_topic_run_parser(subparsers): @@ -131,6 +142,14 @@ def make_topic_run_parser(subparsers): ) topic_parser.add_argument("--message-size", default=100, type=int, help="Topic message size in bytes") + add_topic_tx_options(topic_parser) + topic_parser.add_argument( + "--messages-per-tx", + default=10, + type=int, + help="Messages written/read per transaction (amortizes tx overhead) in the *-topic-tx workloads", + ) + topic_parser.add_argument("--time", default=10, type=int, help="Time to run in seconds") topic_parser.add_argument( "--shutdown-time", @@ -152,6 +171,7 @@ def make_topic_cleanup_parser(subparsers): add_common_options(topic_cleanup_parser) topic_cleanup_parser.add_argument("--path", default="/local/slo_topic", type=str, help="Topic path") + add_topic_tx_options(topic_cleanup_parser) def get_root_parser(): diff --git a/tests/slo/src/runners/topic_runner.py b/tests/slo/src/runners/topic_runner.py index c9a8bdaa..5d717f78 100644 --- a/tests/slo/src/runners/topic_runner.py +++ b/tests/slo/src/runners/topic_runner.py @@ -1,21 +1,56 @@ import time +from os import path -from core.metrics import create_metrics +from core.metrics import WORKLOAD, create_metrics from jobs.async_topic_jobs import AsyncTopicJobManager +from jobs.async_topic_tx_jobs import AsyncTopicTxJobManager from jobs.topic_jobs import TopicJobManager +from jobs.topic_tx_jobs import TopicTxJobManager import ydb import ydb.aio from .base import BaseRunner +SINK_TABLE_DDL = """ +CREATE TABLE IF NOT EXISTS `{}` ( + writer_id Uint64, + seqno Uint64, + write_ts_ns Uint64, + received_at Timestamp, + PRIMARY KEY (writer_id, seqno) +); +""" + class TopicRunner(BaseRunner): @property def prefix(self) -> str: return "topic" + @staticmethod + def _is_tx_workload() -> bool: + return str(WORKLOAD).endswith("topic-tx") + + @staticmethod + def _tx_sink_table(args): + return path.join(args.db, args.sink_table) + def create(self, args): + self._create_topic(args) + if self._is_tx_workload(): + self._create_tx_tables(args) + + def _create_tx_tables(self, args): + sink_table = self._tx_sink_table(args) + self.logger.info("Creating tx sink table: %s", sink_table) + + assert self.driver is not None, "Driver is not initialized. Call set_driver() before create()." + with ydb.QuerySessionPool(self.driver) as pool: + pool.execute_with_retries(SINK_TABLE_DDL.format(sink_table)) + self.logger.info("Tx sink table ready") + + def _create_topic(self, args): assert self.driver is not None, "Driver is not initialized. Call set_driver() before create()." retry_no = 0 while retry_no < 3: @@ -74,7 +109,10 @@ def run(self, args): self.logger.info("Starting topic SLO tests") - job_manager = TopicJobManager(self.driver, args, metrics) + if self._is_tx_workload(): + job_manager = TopicTxJobManager(self.driver, args, metrics, self._tx_sink_table(args)) + else: + job_manager = TopicJobManager(self.driver, args, metrics) job_manager.run_tests() self.logger.info("Topic SLO tests completed") @@ -90,7 +128,10 @@ async def run_async(self, args): self.logger.info("Starting async topic SLO tests") # Use async driver for topic operations - job_manager = AsyncTopicJobManager(self.driver, args, metrics) + if self._is_tx_workload(): + job_manager = AsyncTopicTxJobManager(self.driver, args, metrics, self._tx_sink_table(args)) + else: + job_manager = AsyncTopicJobManager(self.driver, args, metrics) await job_manager.run_tests() self.logger.info("Async topic SLO tests completed") @@ -111,3 +152,12 @@ def cleanup(self, args): else: self.logger.error("Failed to drop topic: %s", e) raise + + if self._is_tx_workload(): + sink_table = self._tx_sink_table(args) + with ydb.QuerySessionPool(self.driver) as pool: + try: + pool.execute_with_retries("DROP TABLE `{}`;".format(sink_table)) + self.logger.info("Table dropped: %s", sink_table) + except ydb.Error as e: + self.logger.info("Table not dropped (%s): %s", sink_table, e)