Merge pull request #311 from ydb-platform/slo

slo: workload through sync ydb driver

Author: Valeria1235

.github/workflows/slo.yml

@@ -0,0 +1,55 @@
+name: SLO
+
+on:
+  pull_request:
+    branches: [main]
+  workflow_dispatch:
+
+jobs:
+  test-slo:
+    concurrency:
+      group: slo-${{ github.ref }}
+    if: (!contains(github.event.pull_request.labels.*.name, 'no slo'))
+
+    runs-on: ubuntu-latest
+    name: SLO test
+    permissions:
+      checks: write
+      pull-requests: write
+      contents: read
+      issues: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+
+      - name: Run SLO
+        uses: ydb-platform/slo-tests@js-version
+        with:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          KUBECONFIG_B64: ${{ secrets.SLO_KUBE_CONFIG }}
+          AWS_CREDENTIALS_B64: ${{ secrets.SLO_AWS_CREDENTIALS }}
+          AWS_CONFIG_B64: ${{ secrets.SLO_AWS_CONFIG }}
+          DOCKER_USERNAME: ${{ secrets.SLO_DOCKER_USERNAME }}
+          DOCKER_PASSWORD: ${{ secrets.SLO_DOCKER_PASSWORD }}
+          DOCKER_REPO: ${{ secrets.SLO_DOCKER_REPO }}
+          DOCKER_FOLDER: ${{ secrets.SLO_DOCKER_FOLDER }}
+          s3_endpoint: ${{ secrets.SLO_S3_ENDPOINT }}
+          s3_images_folder: ${{ vars.SLO_S3_IMAGES_FOLDER }}
+          grafana_domain: ${{ vars.SLO_GRAFANA_DOMAIN }}
+          grafana_dashboard: ${{ vars.SLO_GRAFANA_DASHBOARD }}
+          ydb_version: 'newest'
+          timeBetweenPhases: 30
+          shutdownTime: 30
+
+          language_id0: sync
+          language0: python-sync
+          workload_path0: tests/slo
+          workload_build_context0: ../..
+          workload_build_options0: -f Dockerfile
+
+      - uses: actions/upload-artifact@v3
+        if: always()
+        with:
+          name: slo-logs
+          path: logs/

docker-compose.yml

@@ -1,4 +1,4 @@
-version: "3.9"
+version: "3.3"
 services:
   ydb:
     image: cr.yandex/yc/yandex-docker-local-ydb:latest

tests/slo/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.8
+COPY . /src
+WORKDIR /src
+RUN python -m pip install --upgrade pip && python -m pip install -e . && python -m pip install -r tests/slo/requirements.txt
+WORKDIR tests/slo
+
+ENTRYPOINT ["python", "src"]

tests/slo/README.md

@@ -0,0 +1,133 @@
+# SLO workload
+
+SLO is the type of test where app based on ydb-sdk is tested against falling YDB cluster nodes, tablets, network
+(that is possible situations for distributed DBs with hundreds of nodes)
+
+### Implementations:
+
+There are two implementations:
+
+- `sync`
+- `async` (now unimplemented)
+
+### Usage:
+
+It has 3 commands:
+
+- `create`  - creates table in database
+- `cleanup` - drops table in database
+- `run`     - runs workload (read and write to table with sets RPS)
+
+### Run examples with all arguments:
+
+create:
+`python tests/slo/src/ create localhost:2136 /local -t tableName
+--min-partitions-count 6 --max-partitions-count 1000 --partition-size 1 -с 1000
+--write-timeout 10000`
+
+cleanup:
+`python tests/slo/src/ cleanup localhost:2136 /local -t tableName`
+
+run:
+`python tests/slo/src/ run localhost:2136 /local -t tableName
+--prom-pgw http://prometheus-pushgateway:9091 -report-period 250
+--read-rps 1000 --read-timeout 10000
+--write-rps 100 --write-timeout 10000
+--time 600 --shutdown-time 30`
+
+## Arguments for commands:
+
+### create
+`python tests/slo/src/ create <endpoint> <db> [options]`
+
+```
+Arguments:
+  endpoint                        YDB endpoint to connect to
+  db                              YDB database to connect to
+
+Options:
+  -t --table-name                  <string> table name to create
+
+  -p-min   --min-partitions-count  <int>    minimum amount of partitions in table
+  -p-max   --max-partitions-count  <int>    maximum amount of partitions in table
+  -p-size  --partition-size        <int>    partition size in mb
+
+  -c --initial-data-count          <int>    amount of initially created rows
+
+  --write-timeout                  <int>    write timeout milliseconds
+
+  --batch-size                     <int>    amount of new records in each create request
+  --threads                        <int>    number of threads to use
+
+```
+
+### cleanup
+`python tests/slo/src/ cleanup <endpoint> <db> [options]`
+
+```
+Arguments:
+  endpoint                        YDB endpoint to connect to
+  db                              YDB database to connect to
+
+Options:
+  -t --table-name                  <string> table name to create
+```
+
+### run
+`python tests/slo/src/ run <endpoint> <db> [options]`
+
+```
+Arguments:
+  endpoint                        YDB endpoint to connect to
+  db                              YDB database to connect to
+
+Options:
+  -t --table-name         <string> table name to create
+
+  --prom-pgw              <string> prometheus push gateway
+  --report-period         <int>    prometheus push period in milliseconds
+
+  --read-rps              <int>    read RPS
+  --read-timeout          <int>    read timeout milliseconds
+
+  --write-rps             <int>    write RPS
+  --write-timeout         <int>    write timeout milliseconds
+
+  --time                  <int>    run time in seconds
+  --shutdown-time         <int>    graceful shutdown time in seconds
+
+  --read-threads          <int>    number of threads to use for write requests
+  --write-threads         <int>    number of threads to use for read requests
+```
+
+## Authentication
+
+Workload using [auth-env](https://ydb.yandex-team.ru/docs/reference/ydb-sdk/recipes/auth-env) for authentication.
+
+## What's inside
+When running `run` command, the program creates three jobs: `readJob`, `writeJob`, `metricsJob`.
+
+- `readJob`    reads rows from the table one by one with random identifiers generated by writeJob
+- `writeJob`   generates and inserts rows
+- `metricsJob` periodically sends metrics to Prometheus
+
+Table have these fields:
+- `object_id Uint64`
+- `object_hash Uint64 Digest::NumericHash(id)`
+- `payload_str UTF8`
+- `payload_double Double`
+- `payload_timestamp Timestamp`
+
+Primary key: `("object_hash", "object_id")`
+
+## Collected metrics
+- `oks`      - amount of OK requests
+- `not_oks`  - amount of not OK requests
+- `inflight` - amount of requests in flight
+- `latency`  - summary of latencies in ms
+- `attempts` - summary of amount for request
+
+> You must reset metrics to keep them `0` in prometheus and grafana before beginning and after ending of jobs
+
+## Look at metrics in grafana
+You can get dashboard used in that test [here](https://github.com/ydb-platform/slo-tests/blob/main/k8s/helms/grafana.yaml#L69) - you will need to import json into grafana.

tests/slo/requirements.txt

@@ -0,0 +1,4 @@
+requests==2.28.2
+ratelimiter==1.2.0.post0
+prometheus-client==0.17.0
+quantile-estimator==0.1.2

tests/slo/src/__init__.py

@@ -0,0 +1,13 @@
+import gc
+import logging
+
+from options import parse_options
+from runner import run_from_args
+
+logging.basicConfig(level=logging.INFO)
+
+
+if __name__ == "__main__":
+    args = parse_options()
+    gc.disable()
+    run_from_args(args)

tests/slo/src/__main__.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+import dataclasses
+import logging
+import random
+import string
+from datetime import datetime
+from threading import Lock
+
+logger = logging.getLogger(__name__)
+
+
+MAX_UINT64 = 2**64 - 1
+
+
+def generate_random_string(min_len, max_len):
+    strlen = random.randint(min_len, max_len)
+    return "".join(random.choices(string.ascii_lowercase, k=strlen))
+
+
+@dataclasses.dataclass
+class Row:
+    object_id: int
+    payload_str: str
+    payload_double: float
+    payload_timestamp: datetime
+
+
+@dataclasses.dataclass
+class RowGenerator:
+    id_counter: int = 0
+    lock = Lock()
+
+    def get(self):
+        with self.lock:
+            self.id_counter += 1
+            if self.id_counter >= MAX_UINT64:
+                self.id_counter = 0
+                logger.warning("RowGenerator: maxint reached")
+
+        return Row(
+            object_id=self.id_counter,
+            payload_str=generate_random_string(20, 40),
+            payload_double=random.random(),
+            payload_timestamp=datetime.now(),
+        )
+
+
+def batch_generator(args, start_id=0):
+    row_generator = RowGenerator(start_id)
+    remain = args.initial_data_count
+
+    while True:
+        size = min(remain, args.batch_size)
+        if size < 1:
+            return
+        yield [row_generator.get() for _ in range(size)]
+        remain -= size