Topic multi-partition writer (write-by-key) #1952
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| name: SLO | |
| on: | |
| pull_request: | |
| types: [opened, reopened, synchronize, labeled] | |
| permissions: | |
| contents: read | |
| pull-requests: write | |
| checks: write | |
| jobs: | |
| select-scenarios: | |
| # On `labeled` events run only when the `SLO` label itself was just added — | |
| # otherwise unrelated label changes (e.g. the AI-review bot toggling | |
| # `ai_review_in_process` / `ai_reviewed`) would spawn a fresh run that | |
| # cancels the in-progress one via `cancel-in-progress`. For the other | |
| # trigger types keep gating on the `SLO` label being present. | |
| # | |
| # The gate lives only here: the workload job `needs` this one, so a skipped | |
| # gate skips the whole run exactly as before. | |
| if: >- | |
| (github.event.action == 'labeled' && github.event.label.name == 'SLO') || | |
| (github.event.action != 'labeled' && contains(github.event.pull_request.labels.*.name, 'SLO')) | |
| name: Select SLO scenarios | |
| runs-on: ubuntu-latest | |
| outputs: | |
| sdk: ${{ steps.select.outputs.sdk }} | |
| steps: | |
| # A full SLO run is long and burns external runners, so only the scenarios that can | |
| # actually be affected by the diff are started. The scenario list lives here (and only | |
| # here) so the map and the definitions cannot drift apart. | |
| - name: Pick scenarios affected by the changed files | |
| id: select | |
| env: | |
| GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} | |
| PR_NUMBER: ${{ github.event.pull_request.number }} | |
| REPO: ${{ github.repository }} | |
| run: | | |
| set -euo pipefail | |
| ALL_SCENARIOS='[ | |
| {"name":"sync-table","command":"--read-rps 1000 --write-rps 100"}, | |
| {"name":"sync-query","command":"--read-rps 1000 --write-rps 100"}, | |
| {"name":"async-query","command":"--read-rps 1000 --write-rps 100"}, | |
| {"name":"sync-topic","command":"--write-rps 200 --write-threads 8 --read-threads 8", | |
| "metrics_yaml_path":"sdk-current/tests/slo/metrics-topic.yaml", | |
| "thresholds_yaml_path":"sdk-current/tests/slo/thresholds-topic.yaml"}, | |
| {"name":"async-topic","command":"--write-rps 200 --write-threads 8 --read-threads 8", | |
| "metrics_yaml_path":"sdk-current/tests/slo/metrics-topic.yaml", | |
| "thresholds_yaml_path":"sdk-current/tests/slo/thresholds-topic.yaml"}, | |
| {"name":"sync-topic-multiwriter", | |
| "command":"--write-rps 200 --write-threads 2 --keys-per-writer 16 --read-threads 8", | |
| "metrics_yaml_path":"sdk-current/tests/slo/metrics-topic.yaml", | |
| "thresholds_yaml_path":"sdk-current/tests/slo/thresholds-topic.yaml"} | |
| ]' | |
| CHANGED=$(gh api "repos/${REPO}/pulls/${PR_NUMBER}/files" --paginate --jq '.[].filename') | |
| echo "Changed files:" | |
| echo "${CHANGED}" | sed 's/^/ /' | |
| TOPIC=0; QUERY=0; TABLE=0; EVERYTHING=0 | |
| while IFS= read -r file; do | |
| [ -n "${file}" ] || continue | |
| case "${file}" in | |
| # Order matters: the first matching pattern wins, so the specific service | |
| # paths have to be tested before the `ydb/*` catch-all below. | |
| ydb/_topic_reader/*|ydb/_topic_writer/*|ydb/_topic_common/*| \ | |
| ydb/topic.py|ydb/aio/topic.py|ydb/_grpc/grpcwrapper/ydb_topic*.py) | |
| TOPIC=1 ;; | |
| ydb/query/*|ydb/aio/query/*) | |
| QUERY=1 ;; | |
| ydb/table.py|ydb/aio/table.py|ydb/_session_impl.py) | |
| TABLE=1 ;; | |
| # Harness parts that belong to one service only. | |
| tests/slo/src/jobs/*topic*|tests/slo/src/runners/topic_runner.py| \ | |
| tests/slo/metrics-topic.yaml|tests/slo/thresholds-topic.yaml) | |
| TOPIC=1 ;; | |
| # One runner drives sync-table, sync-query and async-query alike. | |
| tests/slo/src/jobs/*table*|tests/slo/src/runners/table_runner.py) | |
| TABLE=1; QUERY=1 ;; | |
| # Prose about the harness changes no behaviour. | |
| tests/slo/*.md) | |
| ;; | |
| # Shared harness (options, runners entry point, metrics, image) or this workflow: | |
| # every scenario is affected. | |
| tests/slo/*|.github/workflows/slo.yml) | |
| EVERYTHING=1 ;; | |
| # Anything else inside the SDK is shared machinery (driver, pool, connection, | |
| # retries, credentials, generated stubs, convert/types used by more than one | |
| # service) — assume it can move any scenario. | |
| ydb/*) | |
| EVERYTHING=1 ;; | |
| # Docs, examples, packaging, other CI: no SLO impact. | |
| *) ;; | |
| esac | |
| done <<EOF | |
| ${CHANGED} | |
| EOF | |
| WANT="" | |
| if [ "${EVERYTHING}" -eq 1 ]; then | |
| WANT="all" | |
| else | |
| [ "${TABLE}" -eq 1 ] && WANT="${WANT} sync-table" | |
| [ "${QUERY}" -eq 1 ] && WANT="${WANT} sync-query async-query" | |
| [ "${TOPIC}" -eq 1 ] && WANT="${WANT} sync-topic async-topic sync-topic-multiwriter" | |
| fi | |
| # Nothing SLO-relevant changed, yet someone deliberately asked for a run: honour the | |
| # request and run everything. This doubles as the "run all scenarios" escape hatch — | |
| # touch any non-SDK file and add the label. | |
| if [ -z "${WANT# }" ]; then | |
| echo "No SLO-relevant paths changed; running every scenario." | |
| WANT="all" | |
| fi | |
| if [ "${WANT}" = "all" ]; then | |
| SDK=$(printf '%s' "${ALL_SCENARIOS}" | jq -c .) | |
| else | |
| SDK=$(printf '%s' "${ALL_SCENARIOS}" \ | |
| | jq -c --arg want "${WANT# }" '[ .[] | select(.name as $n | ($want | split(" ")) | index($n)) ]') | |
| fi | |
| echo "Selected scenarios: $(printf '%s' "${SDK}" | jq -r '[.[].name] | join(", ")')" | |
| echo "sdk=${SDK}" >> "$GITHUB_OUTPUT" | |
| ydb-slo-action: | |
| needs: select-scenarios | |
| name: Run YDB SLO Tests | |
| runs-on: "large-runner-python-sdk" | |
| strategy: | |
| fail-fast: false | |
| matrix: | |
| sdk: ${{ fromJSON(needs.select-scenarios.outputs.sdk) }} | |
| concurrency: | |
| group: slo-${{ github.ref }}-${{ matrix.sdk.name }} | |
| cancel-in-progress: true | |
| steps: | |
| - name: Install dependencies | |
| run: | | |
| set -euxo pipefail | |
| YQ_VERSION=v4.48.2 | |
| BUILDX_VERSION=0.30.1 | |
| COMPOSE_VERSION=2.40.3 | |
| sudo curl -fLo /usr/local/bin/yq \ | |
| "https://github.com/mikefarah/yq/releases/download/${YQ_VERSION}/yq_linux_amd64" | |
| sudo chmod +x /usr/local/bin/yq | |
| sudo mkdir -p /usr/local/lib/docker/cli-plugins | |
| sudo curl -fLo /usr/local/lib/docker/cli-plugins/docker-buildx \ | |
| "https://github.com/docker/buildx/releases/download/v${BUILDX_VERSION}/buildx-v${BUILDX_VERSION}.linux-amd64" | |
| sudo chmod +x /usr/local/lib/docker/cli-plugins/docker-buildx | |
| sudo curl -fLo /usr/local/lib/docker/cli-plugins/docker-compose \ | |
| "https://github.com/docker/compose/releases/download/v${COMPOSE_VERSION}/docker-compose-linux-x86_64" | |
| sudo chmod +x /usr/local/lib/docker/cli-plugins/docker-compose | |
| yq --version | |
| docker --version | |
| docker buildx version | |
| docker compose version | |
| - name: Checkout current SDK version | |
| uses: actions/checkout@v5 | |
| with: | |
| path: sdk-current | |
| fetch-depth: 0 | |
| - name: Determine baseline commit | |
| id: baseline | |
| working-directory: sdk-current | |
| env: | |
| SLO_SCENARIO: ${{ matrix.sdk.name }} | |
| PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }} | |
| run: | | |
| set -euo pipefail | |
| BASELINE=$(git merge-base HEAD origin/main) | |
| if [ "${SLO_SCENARIO}" = "sync-topic-multiwriter" ] && \ | |
| ! git grep -q -F 'def multiwriter(' "${BASELINE}" -- ydb/topic.py; then | |
| # Before the API reaches main, compare against the preceding SDK implementation. | |
| # Harness-only commits must not turn this into a comparison with identical SDK code. | |
| LAST_IMPLEMENTATION=$(git log -1 --format=%H "${PR_HEAD_SHA}" -- \ | |
| ydb/_topic_writer ydb/topic.py ':(exclude)**/*_test.py') | |
| BASELINE=$(git rev-parse "${LAST_IMPLEMENTATION}^") | |
| if ! git grep -q -F 'def multiwriter(' "${BASELINE}" -- ydb/topic.py; then | |
| echo "No earlier multiwriter implementation is available for comparison." >&2 | |
| exit 1 | |
| fi | |
| BASELINE_REF="previous-multiwriter@${BASELINE:0:7}" | |
| elif git merge-base --is-ancestor "${BASELINE}" origin/main && \ | |
| [ "$(git rev-parse origin/main)" = "${BASELINE}" ]; then | |
| BASELINE_REF="main" | |
| else | |
| BRANCH=$(git branch -r --contains "${BASELINE}" | grep -v HEAD | head -1 | sed 's|.*/||' || echo "") | |
| if [ -n "${BRANCH}" ]; then | |
| BASELINE_REF="${BRANCH}@${BASELINE:0:7}" | |
| else | |
| BASELINE_REF="${BASELINE:0:7}" | |
| fi | |
| fi | |
| echo "sha=${BASELINE}" >> "$GITHUB_OUTPUT" | |
| echo "ref=${BASELINE_REF}" >> "$GITHUB_OUTPUT" | |
| - name: Checkout baseline SDK version | |
| uses: actions/checkout@v5 | |
| with: | |
| ref: ${{ steps.baseline.outputs.sha }} | |
| path: sdk-baseline | |
| fetch-depth: 1 | |
| - name: Build workload images (current + baseline) | |
| run: | | |
| set -euxo pipefail | |
| # Build current: SDK + workload runner both from this PR. | |
| docker build \ | |
| -f "$GITHUB_WORKSPACE/sdk-current/tests/slo/Dockerfile" \ | |
| -t "ydb-app-current" \ | |
| "$GITHUB_WORKSPACE/sdk-current" | |
| # Build baseline: baseline SDK with the current workload runner | |
| # (Dockerfile + tests/slo/), so the runner-side contract changes | |
| # (entrypoint, metrics format) apply uniformly to both images. | |
| rm -rf "$GITHUB_WORKSPACE/sdk-baseline/tests/slo" | |
| cp -r "$GITHUB_WORKSPACE/sdk-current/tests/slo" \ | |
| "$GITHUB_WORKSPACE/sdk-baseline/tests/slo" | |
| docker build \ | |
| -f "$GITHUB_WORKSPACE/sdk-baseline/tests/slo/Dockerfile" \ | |
| -t "ydb-app-baseline" \ | |
| "$GITHUB_WORKSPACE/sdk-baseline" | |
| - name: Run SLO Tests | |
| uses: ydb-platform/ydb-slo-action/init@v2 | |
| timeout-minutes: 30 | |
| with: | |
| github_issue: ${{ github.event.pull_request.number }} | |
| github_token: ${{ secrets.GITHUB_TOKEN }} | |
| workload_name: ${{ matrix.sdk.name }} | |
| workload_duration: "600" | |
| fail_on_workload_error: "true" | |
| workload_current_ref: ${{ github.head_ref || github.ref_name }} | |
| workload_current_image: ydb-app-current | |
| workload_current_command: ${{ matrix.sdk.command }} | |
| workload_baseline_ref: ${{ steps.baseline.outputs.ref }} | |
| workload_baseline_image: ydb-app-baseline | |
| workload_baseline_command: ${{ matrix.sdk.command }} | |
| # Custom metrics (e.g. topic e2e latency / loss) merged on top of the | |
| # action defaults; empty for workloads that don't set it. | |
| metrics_yaml_path: ${{ matrix.sdk.metrics_yaml_path }} | |
| # Per-scenario threshold overrides (topics make read_latency neutral). | |
| # Ignored by action versions without ydb-slo-action#57; self-activates | |
| # once per-scenario thresholds land in the consumed tag. | |
| thresholds_yaml_path: ${{ matrix.sdk.thresholds_yaml_path }} | |
| - name: Inspect workload exit status | |
| if: always() | |
| run: | | |
| docker inspect --format \ | |
| '{{.Name}} status={{.State.Status}} exit={{.State.ExitCode}} oom_killed={{.State.OOMKilled}} memory_limit={{.HostConfig.Memory}} error={{.State.Error}}' \ | |
| ydb-workload-current ydb-workload-baseline |