temp

Author: vgvoleg

tests/slo/slo_runner.sh

@@ -1,9 +1,16 @@
 docker compose -f playground/configs/compose.yaml down -v
 docker compose -f playground/configs/compose.yaml up -d --wait
 
-../../.venv/bin/python ./src topic-create grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic
+EXECUTOR="../../.venv/bin/python ./src"
 
-../../.venv/bin/python ./src topic-run grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic --prom-pgw "" --read-threads 0 --write-rps 1 --time 120
 
-# ../../.venv/bin/python ./src topic-run grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic --prom-pgw "" --read-threads 0 --time 5
-# ../../.venv/bin/python ./src topic-run grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic --prom-pgw "" --write-threads 0 --read-rps 1 --time 200
+# $EXECUTOR topic-create grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic
+# $EXECUTOR topic-run grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic --prom-pgw "" --read-threads 0 --write-rps 1 --time 120
+
+$EXECUTOR topic-create grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic
+$EXECUTOR topic-run grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic --prom-pgw "" --read-threads 0 --write-rps 5 --time 60
+$EXECUTOR topic-run grpc://localhost:2135 /Root/testdb --path /Root/testdb/slo_topic --prom-pgw "" --write-threads 0 --read-rps 1 --time 60
+
+# $EXECUTOR table-create grpc://localhost:2135 /Root/testdb
+# $EXECUTOR table-run grpc://localhost:2135 /Root/testdb --prom-pgw "" --read-rps 10 --write-threads 0 --read-threads 5 --time 600 --session-pool-size 1
+# $EXECUTOR table-cleanup grpc://localhost:2135 /Root/testdb

tests/slo/src/core/metrics.py

@@ -45,14 +45,24 @@ def create_metrics(push_gateway) -> BaseMetrics:
 
 
 class DummyMetrics(BaseMetrics):
+    def __init__(self):
+        self._metrics = dict()
+
     def start(self, labels):
+        for label in labels:
+            if label not in self._metrics:
+                self._metrics[label] = {'success': 0, 'failure': 0}
         return 0
 
     def stop(self, labels, start_time, attempts=1, error=None):
-        pass
+        for label in labels:
+            if error:
+                self._metrics[label]['failure'] += 1
+            else:
+                self._metrics[label]['success'] += 1
 
     def reset(self):
-        pass
+        logger.info(f"Collected metrics: {self._metrics}")
 
 
 class Metrics(BaseMetrics):

tests/slo/src/jobs/table_jobs.py

@@ -65,6 +65,7 @@ def transaction(session):
 
     try:
         params.pool.retry_operation_sync(transaction, retry_settings=params.retry_settings)
+        logger.info(f"Successfully executed query with params: {params.params}")
     except ydb.Error as err:
         error = err
         logger.exception("[labels: %s] Cannot retry error:", params.labels)
@@ -190,7 +191,7 @@ def _run_table_reads(self, query, limiter):
             max_session_acquire_timeout=self.args.read_timeout / 1000,
         )
 
-        with ydb.SessionPool(self.driver) as pool:
+        with ydb.SessionPool(self.driver, size=self.args.session_pool_size) as pool:
             while time.time() - start_time < self.args.time:
                 params = {"$object_id": randint(1, self.max_id)}
 
@@ -225,7 +226,7 @@ def _run_table_writes(self, query, limiter):
 
         row_generator = RowGenerator(self.max_id)
 
-        with ydb.SessionPool(self.driver) as pool:
+        with ydb.SessionPool(self.driver, size=self.args.session_pool_size) as pool:
             while time.time() - start_time < self.args.time:
                 row = row_generator.get()
                 params = {
@@ -259,7 +260,7 @@ def _run_query_reads(self, query, limiter):
             max_session_acquire_timeout=self.args.read_timeout / 1000,
         )
 
-        with ydb.QuerySessionPool(self.driver) as pool:
+        with ydb.QuerySessionPool(self.driver, size=self.args.session_pool_size) as pool:
             while time.time() - start_time < self.args.time:
                 params = {"$object_id": (randint(1, self.max_id), ydb.PrimitiveType.Uint64)}
 
@@ -295,7 +296,7 @@ def _run_query_writes(self, query, limiter):
 
         row_generator = RowGenerator(self.max_id)
 
-        with ydb.QuerySessionPool(self.driver) as pool:
+        with ydb.QuerySessionPool(self.driver, size=self.args.session_pool_size) as pool:
             while time.time() - start_time < self.args.time:
                 row = row_generator.get()
                 params = {

tests/slo/src/options.py

@@ -38,6 +38,7 @@ def make_table_run_parser(subparsers):
     table_run_parser = subparsers.add_parser("table-run", help="Run table SLO workload")
     add_common_options(table_run_parser)
 
+    table_run_parser.add_argument("--session-pool-size", default=100, type=int, help="Session pool size")
     table_run_parser.add_argument("--read-rps", default=100, type=int, help="Read request rps")
     table_run_parser.add_argument(
         "--read-timeout", default=10000, type=int, help="Read requests execution timeout [ms]"

tests/table/table_test.py

@@ -61,3 +61,38 @@ def select(session: ydb.Session):
 
         pool = ydb.SessionPool(driver=driver_sync)
         pool.retry_operation_sync(select)
+
+    def test_async_read_table(self, driver_sync, database):
+        table_path = database + "/test_table_to_read"
+
+        session: ydb.Session = ydb.retry_operation_sync(lambda: driver_sync.table_client.session().create())
+
+        try:
+            session.drop_table(table_path)
+        except ydb.issues.SchemeError:
+            pass
+
+        description = (
+            ydb.TableDescription()
+            .with_primary_keys("key1")
+            .with_columns(
+                ydb.Column("key1", ydb.PrimitiveType.Uint64),
+                ydb.Column("value", ydb.OptionalType(ydb.PrimitiveType.Utf8)),
+            )
+        )
+
+        session.create_table(table_path, description)
+
+        session.transaction().execute(
+            """INSERT INTO `test_table_to_read` (`key1`, `value`) VALUES (1, "hello_world"), (2, "2")""",
+            commit_tx=True,
+        )
+
+        iter = session.async_read_table(table_path)
+        rows_cnt = 0
+        for fut in iter:
+            # item = fut
+            item = fut.result(timeout=60)
+            rows_cnt += len(item.rows)
+
+        assert rows_cnt == 2