ci: migrate to new slo action

Author: polRk

.github/workflows/slo.yml

@@ -1,70 +1,76 @@
-name: SLO
+name: slo
 
 on:
+  push:
+    branches:
+      - main
   pull_request:
-    branches: [main]
+    branches:
+      - main
+      - release-*
   workflow_dispatch:
+    inputs:
+      github_pull_request_number:
+        required: true
+      slo_workload_duration_seconds:
+        default: '600'
+        required: false
+      slo_workload_read_max_rps:
+        default: '1000'
+        required: false
+      slo_workload_write_max_rps:
+        default: '100'
+        required: false
 
 jobs:
-  test-slo:
-    concurrency:
-      group: slo-${{ github.ref }}
+  ydb-slo-action-init:
     if: (!contains(github.event.pull_request.labels.*.name, 'no slo'))
 
+    name: Run YDB SLO Tests
     runs-on: ubuntu-latest
-    name: SLO test
-    permissions:
-      checks: write
-      pull-requests: write
-      contents: read
-      issues: write
+
+    strategy:
+      matrix:
+        sdk:
+          - py-sync-table
+          - py-sync-query
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
-        if: env.DOCKER_REPO != null
-        env:
-          DOCKER_REPO: ${{ secrets.SLO_DOCKER_REPO }}
+        uses: actions/checkout@v4
 
-      - name: Run SLO
-        uses: ydb-platform/slo-tests@main
-        if: env.DOCKER_REPO != null
-        env:
-          DOCKER_REPO: ${{ secrets.SLO_DOCKER_REPO }}
-        continue-on-error: true
+      - name: Install Python3
+        uses: actions/setup-python@v5
         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
+          python-version: '3.8'
+          cache: 'pip'
 
-          language_id0: sync-python-table
-          language0: Python SDK over Table Service
-          workload_path0: tests/slo
-          workload_build_context0: ../..
-          workload_build_options0: -f Dockerfile --build-arg SDK_SERVICE=sync-python-table
+      - name: Install dependencies
+        run: |
+          python -m pip install --no-cache-dir --upgrade pip
+          python -m pip install --no-cache-dir -e .
+          python -m pip install --no-cache-dir -r tests/slo/requirements.txt
 
-          language_id1: sync-python-query
-          language1: Python SDK over Query Service
-          workload_path1: tests/slo
-          workload_build_context1: ../..
-          workload_build_options1: -f Dockerfile --build-arg SDK_SERVICE=sync-python-query
+      - name: Initialize YDB SLO
+        uses: ydb-platform/ydb-slo-action/init@main
+        with:
+          github_pull_request_number: ${{ github.event.inputs.github_pull_request_number }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          sdk_name: ${{ matrix.sdk }}
 
-      - uses: actions/upload-artifact@v3
-        if: env.DOCKER_REPO != null
+      - name: Run SLO Tests
         env:
-          DOCKER_REPO: ${{ secrets.SLO_DOCKER_REPO }}
-        with:
-          name: slo-logs
-          path: logs/
+          REF: '${GITHUB_HEAD_REF:-${GITHUB_REF#refs/heads/}}'
+          SDK_SERVICE: '${{ matrix.sdk }}'
+        run: |
+          python ./tests/slo create grpc://localhost:2135 /Root/testdb
+          python ./tests/slo run grpc://localhost:2135 /Root/testdb \
+            --prom-pgw localhost:9091 \
+            --report-period 250 \
+            --read-rps ${{inputs.slo_workload_read_max_rps || 1000}} \
+            --write-rps ${{inputs.slo_workload_write_max_rps || 100}} \
+            --read-timeout 10000 \
+            --write-timeout 10000 \
+            --time ${{inputs.slo_workload_duration_seconds || 600}} \
+            --shutdown-time 30
+          python ./tests/slo cleanup grpc://localhost:2135 /Root/testdb

tests/slo/Dockerfile

@@ -8,7 +8,7 @@
 
 import threading
 
-from metrics import Metrics, JOB_WRITE_LABEL, JOB_READ_LABEL
+from metrics import Metrics, OP_TYPE_WRITE, OP_TYPE_READ
 
 from generator import RowGenerator
 
@@ -106,7 +106,7 @@ def check_result(result):
                     query=query,
                     params=params,
                     metrics=metrics,
-                    labels=(JOB_READ_LABEL,),
+                    labels=(OP_TYPE_READ,),
                     request_settings=request_settings,
                     retry_settings=retry_setting,
                     check_result_cb=check_result,
@@ -163,7 +163,7 @@ def check_result(result):
                     query=query,
                     params=params,
                     metrics=metrics,
-                    labels=(JOB_READ_LABEL,),
+                    labels=(OP_TYPE_READ,),
                     request_settings=request_settings,
                     retry_settings=retry_setting,
                     check_result_cb=check_result,
@@ -220,7 +220,7 @@ def run_writes(driver, query, row_generator, metrics, limiter, runtime, timeout)
                     query=query,
                     params=params,
                     metrics=metrics,
-                    labels=(JOB_WRITE_LABEL,),
+                    labels=(OP_TYPE_WRITE,),
                     request_settings=request_settings,
                     retry_settings=retry_setting,
                 )
@@ -285,7 +285,7 @@ def check_result(result):
                     query=query,
                     params=params,
                     metrics=metrics,
-                    labels=(JOB_WRITE_LABEL,),
+                    labels=(OP_TYPE_WRITE,),
                     request_settings=request_settings,
                     retry_settings=retry_setting,
                     check_result_cb=check_result,

tests/slo/src/__init__.py renamed to tests/slo/__init__.py

@@ -0,0 +1,158 @@
+import time
+from contextlib import contextmanager
+from importlib.metadata import version
+from collections.abc import Iterable
+
+from os import environ
+
+environ["PROMETHEUS_DISABLE_CREATED_SERIES"] = "True"
+
+from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram, push_to_gateway  # noqa: E402
+
+OP_TYPE_READ, OP_TYPE_WRITE = "read", "write"
+OP_STATUS_SUCCESS, OP_STATUS_FAILURE = "success", "err"
+
+REF = environ.get("REF", "main")
+SDK_SERVICE_NAME = environ.get("SDK_SERVICE", "py-sync-table")
+
+
+class Metrics:
+    def __init__(self, push_gateway):
+        self._push_gtw = push_gateway
+        self._registry = CollectorRegistry()
+        self._metrics = dict(
+            errors_total=Counter(
+                "sdk_errors_total",
+                "Total number of errors encountered, categorized by error type.",
+                labelnames=("operation_type","error_type"),
+                registry=self._registry,
+            ),
+            operations_total=Counter(
+                "sdk_operations_total",
+                "Total number of operations, categorized by type attempted by the SDK.",
+                labelnames=("operation_type",),
+                registry=self._registry,
+            ),
+            operations_success_total=Counter(
+                "sdk_operations_success_total",
+                "Total number of successful operations, categorized by type.",
+                labelnames=("operation_type",),
+                registry=self._registry,
+            ),
+            operations_failure_total=Counter(
+                "sdk_operations_failure_total",
+                "Total number of failed operations, categorized by type.",
+                labelnames=("operation_type",),
+                registry=self._registry,
+            ),
+            operation_latency_seconds=Histogram(
+                "sdk_operation_latency_seconds",
+                "Latency of operations performed by the SDK in seconds, categorized by type and status.",
+                labelnames=("operation_type","operation_status",),
+                registry=self._registry,
+                buckets=(
+                    0.001, # 1 ms
+                    0.002, # 2 ms
+                    0.003, # 3 ms
+                    0.004, # 4 ms
+                    0.005, # 5 ms
+                    0.0075,# 7.5 ms
+                    0.010, # 10 ms
+                    0.020, # 20 ms
+                    0.050, # 50 ms
+                    0.100, # 100 ms
+                    0.200, # 200 ms
+                    0.500, # 500 ms
+                    1.000, # 1 s
+                )
+            ),
+            retry_attempts_total=Counter(
+                "sdk_retry_attempts_total",
+                "Total number of retry attempts, categorized by operation type.",
+                labelnames=("operation_type",),
+                registry=self._registry,
+            ),
+            retries_success_total=Counter(
+                "sdk_retries_success_total",
+                "Total number of successful retries, categorized by operation type.",
+                labelnames=("operation_type",),
+                registry=self._registry,
+            ),
+            retries_failure_total=Counter(
+                "sdk_retries_failure_total",
+                "Total number of failed retries, categorized by operation type.",
+                labelnames=("operation_type",),
+                registry=self._registry,
+            ),
+            pending_operations=Gauge(
+                "sdk_pending_operations",
+                "Current number of pending operations, categorized by type.",
+                labelnames=("operation_type",),
+                registry=self._registry,
+            ),
+        )
+        self.reset()
+
+    def __getattr__(self, item):
+        try:
+            return self._metrics[item]
+        except KeyError:
+            raise AttributeError(f"'{Metrics.__name__}' object has no attribute '{item}'")
+
+    @contextmanager
+    def measure(self, labels):
+        start_ts = self.start(labels)
+        error = None
+        try:
+            yield self
+        except Exception as err:
+            error = err
+        finally:
+            self.stop(labels, start_ts, error=error)
+
+    def start(self, labels):
+        if not isinstance(labels, Iterable):
+            labels = (labels,)
+
+        self.pending_operations.labels(*labels).inc()
+        return time.time()
+
+    def stop(self, labels, start_time, attempts=1, error=None):
+        duration = time.time() - start_time
+
+        if not isinstance(labels, Iterable):
+            labels = (labels,)
+
+        self.operations_total.labels(*labels).inc()
+        self.pending_operations.labels(*labels).dec()
+        self.retry_attempts_total.labels(*labels).inc(attempts)
+
+        if error:
+            self.errors_total.labels(*labels, type(error).__name__).inc()
+            self.retries_failure_total.labels(*labels).inc(attempts)
+            self.operations_failure_total.labels(*labels).inc()
+            self.operation_latency_seconds.labels(*labels, OP_STATUS_FAILURE).observe(duration)
+            return
+
+        self.retries_success_total.labels(*labels).inc(attempts)
+        self.operations_success_total.labels(*labels).inc()
+        self.operation_latency_seconds.labels(*labels, OP_STATUS_SUCCESS).observe(duration)
+
+
+    def push(self):
+        push_to_gateway(
+            self._push_gtw,
+            job=f"workload-{SDK_SERVICE_NAME}",
+            registry=self._registry,
+            grouping_key={
+                "ref": REF,
+                "sdk": SDK_SERVICE_NAME,
+                "sdk_version": version("ydb"),
+            },
+        )
+
+    def reset(self):
+        for m in self._metrics.values():
+            m.clear()
+
+        self.push()

tests/slo/src/__main__.py renamed to tests/slo/__main__.py

@@ -91,13 +91,13 @@ def run_slo(args, driver, tb_name):
     logger.info("Max ID: %s", max_id)
 
     metrics = Metrics(args.prom_pgw)
-    if SDK_SERVICE_NAME == "sync-python-table":
+    if SDK_SERVICE_NAME == "py-sync-table":
         futures = (
             *run_read_jobs(args, driver, tb_name, max_id, metrics),
             *run_write_jobs(args, driver, tb_name, max_id, metrics),
             run_metric_job(args, metrics),
         )
-    elif SDK_SERVICE_NAME == "sync-python-query":
+    elif SDK_SERVICE_NAME == "py-sync-query":
         futures = (
             *run_read_jobs_query(args, driver, tb_name, max_id, metrics),
             *run_write_jobs_query(args, driver, tb_name, max_id, metrics),
@@ -121,7 +121,6 @@ def run_from_args(args):
     driver_config = ydb.DriverConfig(
         args.endpoint,
         database=args.db,
-        credentials=ydb.credentials_from_env_variables(),
         grpc_keep_alive_timeout=5000,
     )