Address SLO review comments

polRk · polRk · commit e8338f410ae2 · 2026-04-25T16:31:56.000+03:00
diff --git a/.github/workflows/slo.yml b/.github/workflows/slo.yml
@@ -137,7 +137,7 @@ jobs:
         uses: ydb-platform/ydb-slo-action/init@v2
         timeout-minutes: 30
         with:
-          github_issue: ${{ github.event.inputs.github_issue }}
+          github_issue: ${{ github.event.pull_request.number || inputs.github_issue }}
           github_token: ${{ secrets.GITHUB_TOKEN }}
           workload_name: ${{ matrix.sdk.name }}
           workload_duration: ${{ inputs.slo_workload_duration_seconds || '600' }}
diff --git a/tests/slo/slo_runner.sh b/tests/slo/slo_runner.sh
@@ -23,6 +23,7 @@ set -euo pipefail
 #   MESSAGE_SIZE    : topic message size bytes (default: 100)
 #   REPORT_PERIOD_MS: metrics flush period ms (default: 1000)
 #   DEBUG           : 1 to enable --debug for workload (default: 0)
+#   ASYNC           : 1 to enable --async for workload (default: 0)
 
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
@@ -39,6 +40,7 @@ WRITE_THREADS="${WRITE_THREADS:-1}"
 MESSAGE_SIZE="${MESSAGE_SIZE:-100}"
 REPORT_PERIOD_MS="${REPORT_PERIOD_MS:-1000}"
 DEBUG="${DEBUG:-0}"
+ASYNC="${ASYNC:-0}"
 
 WORKLOAD_IMAGE="${WORKLOAD_IMAGE:-ydb-python-slo:local}"
 
@@ -78,9 +80,14 @@ build_workload_command() {
     --read-threads "${READ_THREADS}"
     --write-threads "${WRITE_THREADS}"
     --write-rps "${WRITE_RPS}"
-    --message-size "${MESSAGE_SIZE}"
     --time "${RUN_TIME_SEC}"
   )
+  if [[ "${WORKLOAD_NAME}" == "topic" ]]; then
+    cmd+=(--message-size "${MESSAGE_SIZE}")
+    if [[ "${ASYNC}" == "1" ]]; then
+      cmd+=(--async)
+    fi
+  fi
   if [[ "${DEBUG}" == "1" ]]; then
     cmd+=(--debug)
   fi
diff --git a/tests/slo/src/core/metrics.py b/tests/slo/src/core/metrics.py
@@ -132,9 +132,7 @@ def __init__(self, workload_name: str, workload_ref: str):
         resource = Resource.create(
             {
                 "service.name": f"workload-{workload_name}",
-                "service.instance.id": environ.get(
-                    "SLO_INSTANCE_ID", f"{workload_ref}-{workload_name}"
-                ),
+                "service.instance.id": environ.get("SLO_INSTANCE_ID", f"{workload_ref}-{workload_name}"),
                 "ref": workload_ref,
                 "sdk": "ydb-python-sdk",
                 "sdk_version": version("ydb"),
@@ -227,9 +225,7 @@ def stop(
 
         self._retry_attempts_total.add(int(attempts), attributes=base_attrs)
         self._pending.add(-1, attributes=base_attrs)
-        self._operations_total.add(
-            1, attributes={**base_attrs, "operation_status": op_status}
-        )
+        self._operations_total.add(1, attributes={**base_attrs, "operation_status": op_status})
 
         if error is not None:
             self._errors.add(
@@ -291,7 +287,5 @@ def create_metrics(args) -> BaseMetrics:
     try:
         return OtlpMetrics(args.workload_name, args.workload_ref)
     except Exception:
-        logger.exception(
-            "Failed to initialize OTLP metrics — falling back to DummyMetrics"
-        )
+        logger.exception("Failed to initialize OTLP metrics — falling back to DummyMetrics")
         return DummyMetrics()
diff --git a/tests/slo/src/jobs/async_topic_jobs.py b/tests/slo/src/jobs/async_topic_jobs.py
@@ -58,14 +58,12 @@ async def _run_topic_read_jobs(self):
 
         return tasks
 
-    async def _run_topic_writes(
-        self, limiter: AsyncLimiter, partition_id: Optional[int] = None
-    ):
+    async def _run_topic_writes(self, limiter: AsyncLimiter, partition_id: Optional[int] = None):
         start_time = time.time()
         logger.info("Start async topic write workload")
 
         async with self.aio_driver.topic_client.writer(
-            self.args.path,
+            self.args.topic_path,
             codec=getattr(ydb, "PublicCodec", ydb.TopicCodec).GZIP,
             partition_id=partition_id,
         ) as writer:
@@ -100,8 +98,8 @@ async def _run_topic_reads(self, limiter: AsyncLimiter):
         logger.info("Start async topic read workload")
 
         async with self.aio_driver.topic_client.reader(
-            self.args.path,
-            self.args.consumer,
+            self.args.topic_path,
+            self.args.topic_consumer,
         ) as reader:
             logger.info("Async topic reader created")
 
diff --git a/tests/slo/src/jobs/topic_jobs.py b/tests/slo/src/jobs/topic_jobs.py
@@ -66,7 +66,7 @@ def _run_topic_writes(self, limiter, partition_id=None):
         logger.info("Start topic write workload")
 
         with self.driver.topic_client.writer(
-            self.args.path,
+            self.args.topic_path,
             codec=ydb.TopicCodec.GZIP,
             partition_id=partition_id,
         ) as writer:
@@ -100,8 +100,8 @@ def _run_topic_reads(self, limiter):
         logger.info("Start topic read workload")
 
         with self.driver.topic_client.reader(
-            self.args.path,
-            self.args.consumer,
+            self.args.topic_path,
+            self.args.topic_consumer,
         ) as reader:
             logger.info("Topic reader created")
 
diff --git a/tests/slo/src/options.py b/tests/slo/src/options.py
@@ -1,5 +1,50 @@
 import argparse
 import os
+import sys
+
+_WORKLOADS = {"sync-table", "sync-query", "topic"}
+
+_TABLE_OPTIONS = {
+    "--table-name",
+    "--min-partitions-count",
+    "--max-partitions-count",
+    "--partition-size",
+    "--initial-data-count",
+    "--batch-size",
+    "--threads",
+}
+
+_TOPIC_OPTIONS = {
+    "--topic-path",
+    "--topic-consumer",
+    "--topic-partitions",
+    "--message-size",
+}
+
+
+def _provided_options(argv):
+    options = set()
+    for arg in argv:
+        option = arg.split("=", 1)[0]
+        if option.startswith("--"):
+            options.add(option)
+    return options
+
+
+def _validate_options(parser, args, provided_options):
+    if args.workload_name not in _WORKLOADS:
+        parser.error(f"Unknown workload-name: {args.workload_name}. Expected one of: {', '.join(sorted(_WORKLOADS))}")
+
+    if args.workload_name == "topic":
+        invalid_options = sorted(provided_options & _TABLE_OPTIONS)
+    else:
+        invalid_options = sorted(provided_options & _TOPIC_OPTIONS)
+
+    if invalid_options:
+        parser.error(f"{args.workload_name} workload does not accept options: {', '.join(invalid_options)}")
+
+    if args.async_mode and args.workload_name != "topic":
+        parser.error("--async is supported only for topic workload")
 
 
 def parse_options():
@@ -48,7 +93,12 @@ def parse_options():
     )
 
     parser.add_argument("--debug", action="store_true", help="Enable debug logging")
-    parser.add_argument("--async", dest="async_mode", action="store_true", help="Run workload in async mode")
+    parser.add_argument(
+        "--async",
+        dest="async_mode",
+        action="store_true",
+        help="Run workload in async mode",
+    )
 
     # Table params
     parser.add_argument("--table-name", default="key_value", help="Table name")
@@ -81,10 +131,7 @@ def parse_options():
     parser.add_argument("--message-size", default=100, type=int, help="Topic message size [bytes]")
 
     args = parser.parse_args()
-
-    # Aliases used by topic runner
-    args.path = args.topic_path
-    args.consumer = args.topic_consumer
-    args.partitions_count = args.topic_partitions
+    provided_options = _provided_options(sys.argv[1:])
+    _validate_options(parser, args, provided_options)
 
     return args
diff --git a/tests/slo/src/root_runner.py b/tests/slo/src/root_runner.py
@@ -77,13 +77,13 @@ async def _run_all_async(args):
 
         try:
             logger.info("[%s][async] Creating resources", workload_name)
-            runner.create(args)
+            await runner.create_async(args)
 
             logger.info("[%s][async] Running workload for %d s", workload_name, args.time)
             await runner.run_async(args)
         finally:
             logger.info("[%s][async] Cleaning up resources", workload_name)
             try:
-                runner.cleanup(args)
+                await runner.cleanup_async(args)
             except Exception:
                 logger.exception("Cleanup failed — ignoring")
diff --git a/tests/slo/src/runners/base.py b/tests/slo/src/runners/base.py
@@ -22,6 +22,9 @@ def set_driver(self, driver: ydb.Driver):
     def create(self, args):
         pass
 
+    async def create_async(self, args):
+        raise NotImplementedError(f"Async create is not supported for {self.prefix}")
+
     @abstractmethod
     def run(self, args):
         pass
@@ -32,3 +35,6 @@ async def run_async(self, args):
     @abstractmethod
     def cleanup(self, args):
         pass
+
+    async def cleanup_async(self, args):
+        raise NotImplementedError(f"Async cleanup is not supported for {self.prefix}")
diff --git a/tests/slo/src/runners/topic_runner.py b/tests/slo/src/runners/topic_runner.py
@@ -1,3 +1,4 @@
+import asyncio
 import time
 
 from core.metrics import create_metrics
@@ -19,36 +20,42 @@ def create(self, args):
         assert self.driver is not None, "Driver is not initialized. Call set_driver() before create()."
         retry_no = 0
         while retry_no < 3:
-            self.logger.info("Creating topic: %s (retry no: %d)", args.path, retry_no)
+            self.logger.info("Creating topic: %s (retry no: %d)", args.topic_path, retry_no)
 
             try:
                 self.driver.topic_client.create_topic(
-                    path=args.path,
-                    min_active_partitions=args.partitions_count,
-                    max_active_partitions=args.partitions_count,
-                    consumers=[args.consumer],
+                    path=args.topic_path,
+                    min_active_partitions=args.topic_partitions,
+                    max_active_partitions=args.topic_partitions,
+                    consumers=[args.topic_consumer],
                 )
 
-                self.logger.info("Topic created successfully: %s", args.path)
-                self.logger.info("Consumer created: %s", args.consumer)
+                self.logger.info("Topic created successfully: %s", args.topic_path)
+                self.logger.info("Consumer created: %s", args.topic_consumer)
                 return
 
             except ydb.Error as e:
                 error_msg = str(e).lower()
                 if "already exists" in error_msg:
-                    self.logger.info("Topic already exists: %s", args.path)
+                    self.logger.info("Topic already exists: %s", args.topic_path)
 
                     try:
-                        description = self.driver.topic_client.describe_topic(args.path)
-                        consumer_exists = any(c.name == args.consumer for c in description.consumers)
+                        description = self.driver.topic_client.describe_topic(args.topic_path)
+                        consumer_exists = any(c.name == args.topic_consumer for c in description.consumers)
 
                         if not consumer_exists:
-                            self.logger.info("Adding consumer %s to existing topic", args.consumer)
-                            self.driver.topic_client.alter_topic(path=args.path, add_consumers=[args.consumer])
-                            self.logger.info("Consumer added successfully: %s", args.consumer)
+                            self.logger.info(
+                                "Adding consumer %s to existing topic",
+                                args.topic_consumer,
+                            )
+                            self.driver.topic_client.alter_topic(
+                                path=args.topic_path,
+                                add_consumers=[args.topic_consumer],
+                            )
+                            self.logger.info("Consumer added successfully: %s", args.topic_consumer)
                             return
                         else:
-                            self.logger.info("Consumer already exists: %s", args.consumer)
+                            self.logger.info("Consumer already exists: %s", args.topic_consumer)
                             return
 
                     except Exception as alter_err:
@@ -68,6 +75,65 @@ def create(self, args):
 
         raise RuntimeError("Failed to create topic")
 
+    async def create_async(self, args):
+        assert self.driver is not None, "Driver is not initialized. Call set_driver() before create_async()."
+        retry_no = 0
+        while retry_no < 3:
+            self.logger.info("Creating topic: %s (retry no: %d)", args.topic_path, retry_no)
+
+            try:
+                await self.driver.topic_client.create_topic(
+                    path=args.topic_path,
+                    min_active_partitions=args.topic_partitions,
+                    max_active_partitions=args.topic_partitions,
+                    consumers=[args.topic_consumer],
+                )
+
+                self.logger.info("Topic created successfully: %s", args.topic_path)
+                self.logger.info("Consumer created: %s", args.topic_consumer)
+                return
+
+            except ydb.Error as e:
+                error_msg = str(e).lower()
+                if "already exists" in error_msg:
+                    self.logger.info("Topic already exists: %s", args.topic_path)
+
+                    try:
+                        description = await self.driver.topic_client.describe_topic(args.topic_path)
+                        consumer_exists = any(c.name == args.topic_consumer for c in description.consumers)
+
+                        if not consumer_exists:
+                            self.logger.info(
+                                "Adding consumer %s to existing topic",
+                                args.topic_consumer,
+                            )
+                            await self.driver.topic_client.alter_topic(
+                                path=args.topic_path,
+                                add_consumers=[args.topic_consumer],
+                            )
+                            self.logger.info("Consumer added successfully: %s", args.topic_consumer)
+                            return
+                        else:
+                            self.logger.info("Consumer already exists: %s", args.topic_consumer)
+                            return
+
+                    except Exception as alter_err:
+                        self.logger.warning("Failed to add consumer: %s", alter_err)
+                        raise
+                elif "storage pool" in error_msg or "pq" in error_msg:
+                    self.logger.error("YDB instance does not support topics (PersistentQueues): %s", e)
+                    self.logger.error("Please use YDB instance with topic support")
+                    raise
+                elif isinstance(e, ydb.Unavailable):
+                    self.logger.info("YDB instance is not ready, retrying in 5 seconds...")
+                    await asyncio.sleep(5)
+                    retry_no += 1
+                else:
+                    self.logger.error("Failed to create topic: %s", e)
+                    raise
+
+        raise RuntimeError("Failed to create topic")
+
     def run(self, args):
         assert self.driver is not None, "Driver is not initialized. Call set_driver() before run()."
         metrics = create_metrics(args)
@@ -99,15 +165,29 @@ async def run_async(self, args):
             metrics.reset()
 
     def cleanup(self, args):
-        self.logger.info("Cleaning up topic: %s", args.path)
+        self.logger.info("Cleaning up topic: %s", args.topic_path)
 
         assert self.driver is not None, "Driver is not initialized. Call set_driver() before cleanup()."
         try:
-            self.driver.topic_client.drop_topic(args.path)
-            self.logger.info("Topic dropped: %s", args.path)
+            self.driver.topic_client.drop_topic(args.topic_path)
+            self.logger.info("Topic dropped: %s", args.topic_path)
+        except ydb.Error as e:
+            if "does not exist" in str(e).lower():
+                self.logger.info("Topic does not exist: %s", args.topic_path)
+            else:
+                self.logger.error("Failed to drop topic: %s", e)
+                raise
+
+    async def cleanup_async(self, args):
+        self.logger.info("Cleaning up topic: %s", args.topic_path)
+
+        assert self.driver is not None, "Driver is not initialized. Call set_driver() before cleanup_async()."
+        try:
+            await self.driver.topic_client.drop_topic(args.topic_path)
+            self.logger.info("Topic dropped: %s", args.topic_path)
         except ydb.Error as e:
             if "does not exist" in str(e).lower():
-                self.logger.info("Topic does not exist: %s", args.path)
+                self.logger.info("Topic does not exist: %s", args.topic_path)
             else:
                 self.logger.error("Failed to drop topic: %s", e)
                 raise
