Topic workload SLO

Author: vgvoleg

.github/workflows/slo.yml

@@ -30,9 +30,12 @@ jobs:
 
     strategy:
       matrix:
-        workload:
-          - sync-table
-          - sync-query
+        include:
+          - prefix: table
+            workload: sync-table
+          - prefix: table
+            workload: sync-query
+
 
     concurrency:
       group: slo-${{ github.ref }}-${{ matrix.workload }}
@@ -64,14 +67,14 @@ jobs:
 
       - name: Prepare SLO Database
         run: |
-          python ./tests/slo/src create grpc://localhost:2135 /Root/testdb
+          python ./tests/slo/src ${{ matrix.prefix }}-create grpc://localhost:2135 /Root/testdb
 
       - name: Run SLO Tests
         env:
           REF: '${{ github.head_ref || github.ref }}'
           WORKLOAD: '${{ matrix.workload }}'
         run: |
-          python ./tests/slo/src run grpc://localhost:2135 /Root/testdb \
+          python ./tests/slo/src ${{ matrix.prefix }}-run grpc://localhost:2135 /Root/testdb \
             --prom-pgw localhost:9091 \
             --report-period 250 \
             --time ${{inputs.slo_workload_duration_seconds || 600}} \
@@ -83,7 +86,7 @@ jobs:
       - if: always()
         name: Cleanup SLO Database
         run: |
-          python ./tests/slo/src cleanup grpc://localhost:2135 /Root/testdb
+          python ./tests/slo/src ${{ matrix.prefix }}-cleanup grpc://localhost:2135 /Root/testdb
 
       - if: always()
         name: Store ydb chaos testing logs

tests/slo/TOPIC_SLO.md

@@ -0,0 +1,173 @@
+# YDB Python SDK Topic SLO Tests
+
+Этот документ описывает как использовать SLO тесты для топиков YDB.
+
+## Описание
+
+Topic SLO тесты измеряют производительность операций с топиками YDB:
+- **TopicWriter** - запись сообщений в топик
+- **TopicReader** - чтение сообщений из топика с коммитом
+
+Тесты собирают метрики:
+- Latency (задержка операций)
+- RPS (количество операций в секунду)
+- Error rate (процент ошибок)
+- Retry attempts (количество попыток повтора)
+
+## Использование
+
+### Запуск Topic SLO тестов
+
+Topic SLO тесты автоматически создают топик перед началом тестирования и удаляют его после завершения:
+
+```bash
+cd tests/slo
+python -m src topic-run grpc://localhost:2135 /local
+```
+
+### Параметры
+
+#### Основные параметры:
+- `endpoint` - YDB endpoint (например: `grpc://localhost:2135`)
+- `db` - база данных (например: `/local`)
+
+#### Параметры для топиков:
+- `--topic-path` - путь к топику (по умолчанию: `/local/slo_topic`)
+- `--topic-consumer` - имя консьюмера (по умолчанию: `slo_consumer`)
+- `--topic-message-size` - размер сообщения в байтах (по умолчанию: 100)
+
+#### Параметры производительности:
+- `--topic-write-rps` - RPS для записи (по умолчанию: 20)
+- `--topic-read-rps` - RPS для чтения (по умолчанию: 50)
+- `--topic-write-threads` - количество потоков записи (по умолчанию: 2)
+- `--topic-read-threads` - количество потоков чтения (по умолчанию: 4)
+
+#### Таймауты:
+- `--topic-write-timeout` - таймаут записи в мс (по умолчанию: 10000)
+- `--topic-read-timeout` - таймаут чтения в мс (по умолчанию: 5000)
+
+#### Временные параметры:
+- `--time` - время работы теста в секундах (по умолчанию: 60)
+- `--shutdown-time` - время на graceful shutdown в секундах (по умолчанию: 10)
+
+#### Метрики:
+- `--prom-pgw` - Prometheus push gateway (по умолчанию: `localhost:9091`)
+- `--report-period` - период отправки метрик в мс (по умолчанию: 1000)
+
+### Примеры использования
+
+#### Базовый запуск с настройками по умолчанию:
+```bash
+python -m src topic-run grpc://localhost:2135 /local
+```
+
+#### Запуск с повышенной нагрузкой:
+```bash
+python -m src topic-run grpc://localhost:2135 /local \
+  --topic-write-rps 100 \
+  --topic-read-rps 200 \
+  --topic-write-threads 4 \
+  --topic-read-threads 8 \
+  --time 300
+```
+
+#### Запуск с большими сообщениями:
+```bash
+python -m src topic-run grpc://localhost:2135 /local \
+  --topic-message-size 1024 \
+  --topic-write-rps 10 \
+  --topic-read-rps 20
+```
+
+#### Запуск с кастомным топиком и консьюмером:
+```bash
+python -m src topic-run grpc://localhost:2135 /local \
+  --topic-path /local/my_slo_topic \
+  --topic-consumer my_consumer
+```
+
+## Архитектура
+
+### Компоненты
+
+1. **topic_jobs.py** - основная логика для работы с топиками:
+   - `run_topic_writes()` - цикл записи сообщений
+   - `run_topic_reads()` - цикл чтения сообщений
+   - `setup_topic()` - создание топика и консьюмера
+   - `cleanup_topic()` - очистка топика
+
+2. **metrics.py** - расширен для поддержки топик метрик:
+   - `OP_TYPE_TOPIC_WRITE` - метрики записи
+   - `OP_TYPE_TOPIC_READ` - метрики чтения
+
+3. **options.py** - добавлена команда `topic-run` с параметрами для топиков
+
+4. **runner.py** - добавлена функция `run_topic_slo()` для запуска топик тестов
+
+### Workflow
+
+1. **Инициализация**: создание топика и консьюмера (если не существуют)
+2. **Запуск воркеров**:
+   - Потоки записи создают и отправляют сообщения
+   - Потоки чтения получают и коммитят сообщения
+3. **Сбор метрик**: все операции измеряются и отправляются в Prometheus
+4. **Завершение**: graceful shutdown всех воркеров
+
+## Метрики
+
+Топик SLO тесты создают следующие метрики:
+
+### Counters
+- `sdk_operations_total{operation_type="topic_write"}` - общее количество операций записи
+- `sdk_operations_total{operation_type="topic_read"}` - общее количество операций чтения
+- `sdk_operations_success_total{operation_type="topic_write|topic_read"}` - успешные операции
+- `sdk_operations_failure_total{operation_type="topic_write|topic_read"}` - неуспешные операции
+- `sdk_errors_total{operation_type="topic_write|topic_read", error_type="ErrorType"}` - ошибки по типам
+
+### Histograms
+- `sdk_operation_latency_seconds{operation_type="topic_write|topic_read", operation_status="success|err"}` - латентность операций
+
+### Gauges
+- `sdk_pending_operations{operation_type="topic_write|topic_read"}` - количество операций в процессе
+- `sdk_retry_attempts{operation_type="topic_write|topic_read"}` - количество попыток повтора
+
+## Мониторинг
+
+Для мониторинга рекомендуется использовать Grafana с Prometheus. Примеры запросов:
+
+### RPS записи в топик:
+```promql
+rate(sdk_operations_total{operation_type="topic_write"}[1m])
+```
+
+### RPS чтения из топика:
+```promql
+rate(sdk_operations_total{operation_type="topic_read"}[1m])
+```
+
+### Процент ошибок записи:
+```promql
+rate(sdk_operations_failure_total{operation_type="topic_write"}[1m]) /
+rate(sdk_operations_total{operation_type="topic_write"}[1m]) * 100
+```
+
+### 95-й перцентиль латентности чтения:
+```promql
+histogram_quantile(0.95, rate(sdk_operation_latency_seconds_bucket{operation_type="topic_read", operation_status="success"}[1m]))
+```
+
+## Troubleshooting
+
+### Часто встречающиеся проблемы:
+
+1. **Топик не создается**: проверьте права доступа и корректность пути к топику
+2. **Таймауты чтения**: нормально, если нет новых сообщений; увеличьте `--topic-read-timeout` если нужно
+3. **Высокий error rate**: проверьте подключение к YDB и лимиты топика
+4. **Низкий RPS**: увеличьте количество потоков или RPS лимиты
+
+### Логи:
+Тесты логируют в уровне INFO основные события. Для подробной диагностики включите DEBUG:
+```bash
+export PYTHONPATH=src
+python -c "import logging; logging.basicConfig(level=logging.DEBUG)" -m src topic-run ...
+```

tests/slo/src/metrics.py

@@ -10,6 +10,7 @@
 from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram, push_to_gateway  # noqa: E402
 
 OP_TYPE_READ, OP_TYPE_WRITE = "read", "write"
+OP_TYPE_TOPIC_READ, OP_TYPE_TOPIC_WRITE = "topic_read", "topic_write"
 OP_STATUS_SUCCESS, OP_STATUS_FAILURE = "success", "err"
 
 REF = environ.get("REF", "main")

tests/slo/src/options.py

@@ -7,52 +7,96 @@ def add_common_options(parser):
     parser.add_argument("-t", "--table-name", default="key_value", help="Table name")
 
 
-def make_create_parser(subparsers):
-    create_parser = subparsers.add_parser("create", help="Create tables and fill with initial content")
-    add_common_options(create_parser)
+def make_table_create_parser(subparsers):
+    table_create_parser = subparsers.add_parser("table-create", help="Create tables and fill with initial content")
+    add_common_options(table_create_parser)
 
-    create_parser.add_argument(
+    table_create_parser.add_argument(
         "-p-min", "--min-partitions-count", default=6, type=int, help="Minimum amount of partitions in table"
     )
-    create_parser.add_argument(
+    table_create_parser.add_argument(
         "-p-max", "--max-partitions-count", default=1000, type=int, help="Maximum amount of partitions in table"
     )
-    create_parser.add_argument("-p-size", "--partition-size", default=100, type=int, help="Partition size [mb]")
-    create_parser.add_argument(
+    table_create_parser.add_argument("-p-size", "--partition-size", default=100, type=int, help="Partition size [mb]")
+    table_create_parser.add_argument(
         "-c", "--initial-data-count", default=1000, type=int, help="Total number of records to generate"
     )
 
-    create_parser.add_argument("--write-timeout", default=20000, type=int, help="Write requests execution timeout [ms]")
+    table_create_parser.add_argument("--write-timeout", default=20000, type=int, help="Write requests execution timeout [ms]")
 
-    create_parser.add_argument(
+    table_create_parser.add_argument(
         "--batch-size", default=100, type=int, help="Number of new records in each create request"
     )
-    create_parser.add_argument("--threads", default=10, type=int, help="Number of threads to use")
+    table_create_parser.add_argument("--threads", default=10, type=int, help="Number of threads to use")
 
 
-def make_run_parser(subparsers, name="run"):
-    run_parser = subparsers.add_parser(name, help="Run measurable workload")
-    add_common_options(run_parser)
+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)
 
-    run_parser.add_argument("--read-rps", default=100, type=int, help="Read request rps")
-    run_parser.add_argument("--read-timeout", default=10000, type=int, help="Read requests execution timeout [ms]")
+    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]")
 
-    run_parser.add_argument("--write-rps", default=10, type=int, help="Write request rps")
-    run_parser.add_argument("--write-timeout", default=20000, type=int, help="Write requests execution timeout [ms]")
+    table_run_parser.add_argument("--write-rps", default=10, type=int, help="Write request rps")
+    table_run_parser.add_argument("--write-timeout", default=20000, type=int, help="Write requests execution timeout [ms]")
 
-    run_parser.add_argument("--time", default=10, type=int, help="Time to run in seconds")
-    run_parser.add_argument("--shutdown-time", default=10, type=int, help="Graceful shutdown time in seconds")
+    table_run_parser.add_argument("--time", default=10, type=int, help="Time to run in seconds")
+    table_run_parser.add_argument("--shutdown-time", default=10, type=int, help="Graceful shutdown time in seconds")
 
-    run_parser.add_argument("--prom-pgw", default="localhost:9091", type=str, help="Prometheus push gateway")
-    run_parser.add_argument("--report-period", default=1000, type=int, help="Prometheus push period in [ms]")
+    table_run_parser.add_argument("--prom-pgw", default="localhost:9091", type=str, help="Prometheus push gateway")
+    table_run_parser.add_argument("--report-period", default=1000, type=int, help="Prometheus push period in [ms]")
 
-    run_parser.add_argument("--read-threads", default=8, type=int, help="Number of threads to use for write")
-    run_parser.add_argument("--write-threads", default=4, type=int, help="Number of threads to use for read")
+    table_run_parser.add_argument("--read-threads", default=8, type=int, help="Number of threads to use for write")
+    table_run_parser.add_argument("--write-threads", default=4, type=int, help="Number of threads to use for read")
 
 
-def make_cleanup_parser(subparsers):
-    cleanup_parser = subparsers.add_parser("cleanup", help="Drop tables")
-    add_common_options(cleanup_parser)
+def make_topic_run_parser(subparsers):
+    """Создает парсер для команды topic-run - запуск SLO тестов для топиков"""
+    topic_parser = subparsers.add_parser("topic-run", help="Run topic SLO workload")
+    add_common_options(topic_parser)
+
+    topic_parser.add_argument("--topic-read-rps", default=50, type=int, help="Topic read request rps")
+    topic_parser.add_argument("--topic-read-timeout", default=5000, type=int, help="Topic read timeout [ms]")
+    topic_parser.add_argument("--topic-write-rps", default=20, type=int, help="Topic write request rps")
+    topic_parser.add_argument("--topic-write-timeout", default=10000, type=int, help="Topic write timeout [ms]")
+    topic_parser.add_argument("--topic-read-threads", default=1, type=int, help="Number of threads for topic reading")
+    topic_parser.add_argument("--topic-write-threads", default=1, type=int, help="Number of threads for topic writing")
+    topic_parser.add_argument("--topic-path", default="/local/slo_topic", type=str, help="Topic path")
+    topic_parser.add_argument("--topic-consumer", default="slo_consumer", type=str, help="Topic consumer name")
+    topic_parser.add_argument("--topic-message-size", default=100, type=int, help="Topic message size in bytes")
+    topic_parser.add_argument("--topic-min-partitions", default=1, type=int, help="Minimum active partitions")
+    topic_parser.add_argument("--topic-max-partitions", default=10, type=int, help="Maximum active partitions")
+    topic_parser.add_argument("--topic-retention-hours", default=24, type=int, help="Retention period in hours")
+
+    topic_parser.add_argument("--time", default=60, type=int, help="Time to run in seconds")
+    topic_parser.add_argument("--shutdown-time", default=10, type=int, help="Graceful shutdown time in seconds")
+    topic_parser.add_argument("--prom-pgw", default="", type=str, help="Prometheus push gateway (empty to disable)")
+    topic_parser.add_argument("--report-period", default=1000, type=int, help="Prometheus push period in [ms]")
+
+
+def make_topic_create_parser(subparsers):
+    """Создает парсер для команды topic-create"""
+    topic_create_parser = subparsers.add_parser("topic-create", help="Create topic with consumer")
+    add_common_options(topic_create_parser)
+
+    topic_create_parser.add_argument("--topic-path", default="/local/slo_topic", type=str, help="Topic path")
+    topic_create_parser.add_argument("--topic-consumer", default="slo_consumer", type=str, help="Topic consumer name")
+    topic_create_parser.add_argument("--topic-min-partitions", default=1, type=int, help="Minimum active partitions")
+    topic_create_parser.add_argument("--topic-max-partitions", default=10, type=int, help="Maximum active partitions")
+    topic_create_parser.add_argument("--topic-retention-hours", default=24, type=int, help="Retention period in hours")
+
+
+def make_table_cleanup_parser(subparsers):
+    table_cleanup_parser = subparsers.add_parser("table-cleanup", help="Drop tables")
+    add_common_options(table_cleanup_parser)
+
+
+def make_topic_cleanup_parser(subparsers):
+    """Создает парсер для команды topic-cleanup"""
+    topic_cleanup_parser = subparsers.add_parser("topic-cleanup", help="Drop topic")
+    add_common_options(topic_cleanup_parser)
+
+    topic_cleanup_parser.add_argument("--topic-path", default="/local/slo_topic", type=str, help="Topic path")
 
 
 def get_root_parser():
@@ -67,9 +111,12 @@ def get_root_parser():
         help="List of subcommands",
     )
 
-    make_create_parser(subparsers)
-    make_run_parser(subparsers)
-    make_cleanup_parser(subparsers)
+    make_table_create_parser(subparsers)
+    make_table_run_parser(subparsers)
+    make_table_cleanup_parser(subparsers)
+    make_topic_create_parser(subparsers)
+    make_topic_run_parser(subparsers)
+    make_topic_cleanup_parser(subparsers)
 
     return parser