diff --git a/.github/workflows/benchmark-report.yml b/.github/workflows/benchmark-report.yml new file mode 100644 index 00000000..7fabe81e --- /dev/null +++ b/.github/workflows/benchmark-report.yml @@ -0,0 +1,21 @@ +name: Python SDK benchmark report + +on: + workflow_run: + workflows: ['Python SDK benchmarks'] + types: [completed] + +permissions: {} + +jobs: + report: + permissions: + contents: read + actions: read + pull-requests: write + uses: PostHog/.github/.github/workflows/sdk-benchmark-report.yml@3322f26f21f088ad5e5960d65a916971e309ac0b # SDK benchmark workflow (PostHog/.github PR #84) + with: + workflow-path: .github/workflows/benchmarks.yml + artifact-name: python-benchmark-results + report-key: posthog-python + sdk: posthog-python diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index eecd9849..97b76966 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -1,14 +1,6 @@ name: Python SDK benchmarks -on: - pull_request: - paths: - - 'posthog/**' - - 'benchmarks/**' - - 'pyproject.toml' - - 'setup.py' - - 'uv.lock' - - '.github/workflows/benchmarks.yml' +on: [pull_request] permissions: {} @@ -19,7 +11,11 @@ concurrency: jobs: benchmark: runs-on: ubuntu-24.04 - timeout-minutes: 30 + timeout-minutes: 10 + strategy: + fail-fast: false + matrix: + shard: [flags-single, flags-scaling, flags-memory, capture] permissions: contents: read steps: @@ -61,12 +57,67 @@ jobs: run: .venv-benchmark/bin/python -B benchmarks/test_benchmarks.py - name: Measure paired base and head + id: measure + timeout-minutes: 8 run: | .venv-benchmark/bin/python -B benchmarks/run.py \ --base "$BASE_WORKTREE" \ --head "$GITHUB_WORKSPACE" \ --output "$RESULTS" \ - --repetitions 6 + --repetitions 6 \ + --shard "${{ matrix.shard }}" + + - name: Upload raw shard results + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: benchmark-shard-${{ matrix.shard }} + path: | + ${{ runner.temp }}/python-benchmark-results/base.json + ${{ runner.temp }}/python-benchmark-results/head.json + ${{ runner.temp }}/python-benchmark-results/revisions.json + if-no-files-found: warn + retention-days: 14 + + collect: + needs: benchmark + if: ${{ !cancelled() }} + runs-on: ubuntu-24.04 + timeout-minutes: 5 + permissions: + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + + - name: Download shard results + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + pattern: benchmark-shard-* + path: ${{ runner.temp }}/python-benchmark-shards + merge-multiple: false + + - name: Collect and render benchmark comparison + uses: PostHog/.github/.github/actions/benchmark-report@3322f26f21f088ad5e5960d65a916971e309ac0b # SDK benchmark reporting (PostHog/.github PR #84) + with: + mode: collect + source-status: ${{ needs.benchmark.result }} + shard-manifest: benchmarks/shards.json + shard-directory: ${{ runner.temp }}/python-benchmark-shards + output-directory: ${{ runner.temp }}/python-benchmark-report + + - name: Upload rendered comparison + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: python-benchmark-report + path: | + ${{ runner.temp }}/python-benchmark-report/report.md + ${{ runner.temp }}/python-benchmark-report/comparison.json + if-no-files-found: ignore + retention-days: 14 - name: Upload raw benchmark results if: always() @@ -74,7 +125,9 @@ jobs: uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: python-benchmark-results - path: ${{ runner.temp }}/python-benchmark-results + path: | + ${{ runner.temp }}/python-benchmark-report/bundle.json + ${{ runner.temp }}/python-benchmark-report/revisions.json if-no-files-found: warn retention-days: 14 diff --git a/benchmarks/README.md b/benchmarks/README.md index 8ca37a75..d428c862 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -1,6 +1,6 @@ # Python SDK benchmarks -PR-only, paired measurements of independently installed Python SDK revisions. Each workload/pass/repetition runs in a fresh process against the target's non-editable wheel. Benchmark code comes from one checkout and emits `base.json` and `head.json` in the Node pilot's result structure. +PR-only, paired measurements of independently installed Python SDK revisions. Each workload/pass/repetition runs in a fresh process against the target's non-editable wheel. Benchmark code comes from one checkout and emits `base.json` and `head.json` using the shared reporting contract. Suite version 2 implements [sdk-benchmarks/v1](https://github.com/PostHog/.github/blob/main/.github/workflows/sdk-benchmark-specification.md). ## Setup and local execution @@ -26,7 +26,7 @@ Run the controls and benchmarks using the head environment: --repetitions 6 ``` -`--smoke --repetitions 1` checks integration quickly; smoke results are not suitable for performance conclusions. `--filter flags.all.` selects workload IDs by prefix. Using the same revision on both sides measures repeatability, not an improvement. +`--smoke --repetitions 1` checks integration quickly; smoke results are not suitable for performance conclusions. `--filter flags.all.` selects workload IDs by prefix. `--shard flags-single` selects one fixed group from `shards.json`: `flags-single`, `flags-scaling`, `flags-memory`, or `capture`. Shard selection and prefix filtering are mutually exclusive; omitting both runs the full inventory. The manifest must partition both full and smoke inventories exactly, without duplicate workloads. Each invocation creates a fresh pair ID for its own base/head documents. Using the same revision on both sides measures repeatability, not an improvement. Rerun preparation after changing SDK source. Preparation forces a `posthog` rebuild/reinstall so source-only changes cannot reuse uv's cached wheel. The worker deliberately does not import the repository source tree. Results fingerprint the installed package's Python files and record its entry path, SDK commit/dirty state, package version, lockfile digest, actual installed dependency/tool versions, benchmark source digest, runtime, fixture digest, and paired-run ID. @@ -44,37 +44,40 @@ uv pip compile benchmarks/requirements.in --python-version 3.12 --universal \ | Nine single-flag cases | `get_feature_flag` | Elapsed time/op, process CPU/op, Python allocator bytes/calls per op | | All flags: 10, 100, 1,000 | `get_all_flags` | Same metrics per complete all-flags call | | Sustained 1,000-flag evaluation | Eight threads calling `get_all_flags` | RSS observations and post-settling RSS | -| Forty definition refreshes | `load_feature_flags`, then validated `get_all_flags` | Memory checkpoints every five cycles and at the end | +| Forty definition refreshes | `load_feature_flags` | Memory checkpoints after every refresh | | Capture: 32-byte and 4,096-byte string properties | `capture` | Foreground elapsed time and caller-thread CPU per event | -| Capture and drain: both payload sizes | `capture`, `flush` every 100 events | Full-pipeline time, process CPU, allocator bytes/calls, and memory | +| Steady delivery: both payload sizes | `capture`, `flush` every 100 events | Full-pipeline time, process CPU, allocator bytes/calls, and memory | +| Burst delivery: both payload sizes | `capture`, one final `flush` | Full-pipeline time, process CPU, allocator bytes/calls, and memory | -The flag mix and sixteen identities match the Node pilot conceptually: boolean, percentage rollout, multivariate assignment, person properties, regex, cohort, early/late/no-match, and inactive flags. Expected values are computed independently of the SDK. Every identity is validated before and after measurement. Refresh generations alter actual boolean/cohort results and regex patterns so a fetch without a state update fails validation. +The 20 workload IDs and operation counts follow the specification. Full single-flag workloads use 16,000 evaluations; all-flags workloads use 1,600/160/160 for 10/100/1,000 definitions. Allocation passes use 2,000 single-flag and 160/16/16 all-flags evaluations, with the actual denominator recorded. Load uses 1,600 evaluations; refresh uses 40 reloads. Every capture workload and pass uses 1,000 events. Smoke uses 160 single-flag, 16 all-flags, 32 load evaluations, four refreshes, and 200 captures. + +The flag mix and sixteen identities follow the shared fixtures: boolean, percentage rollout, multivariate assignment, person properties, regex, cohort, early/late/no-match, and inactive flags. Expected values are computed independently of the SDK. Every identity is validated before and after measurement. All measured flag phases start at generation 0, restored through public reload after warmup. Refreshes load generations 1–40 (1–4 for smoke), checking each definitions request. Full result sweeps occur before and after measurement, not inside the refresh loop. Controls separately check every changing generation and reload result. The suite covers the supported legacy flag getters, not `evaluate_flags` snapshots. Deprecation warnings are encountered during setup. Eight threads are **GIL-constrained concurrency**, not a claim of eight-way parallel Python execution. ### Transport and lifecycle -Python has no public fetch adapter. A bounded loopback HTTP server runs in the coordinator process, outside the measured worker, and serves definitions and accepts uncompressed v0 batches. Its routes are limited to the fixture protocol and a control/inspection interface. It retains counters and per-identity counts, not event history. Remote flag fallback is unexpected and fails verification. +Python has no public fetch adapter. A bounded loopback HTTP server runs in the coordinator process, outside the measured worker, and serves definitions and accepts uncompressed v0 batches. Its routes are limited to the fixture protocol and a control/inspection interface. It retains bounded counters, per-identity counts, and a preallocated sequence bitmap, not event history. Remote flag fallback is unexpected and fails verification. The SDK uses real requests, JSON parsing/serialization, background consumers, flag loading, flush, and shutdown. No private queue, evaluator, or poller is replaced. Automatic polling and partial-batch deadlines are set to one hour, longer than the worker timeout. Refreshes and draining use public methods. Worker shutdown must stop SDK threads. -Capture checks require the correct event name, exact payload, identity distribution, and total delivered count. Rejected captures, timeouts that leave undelivered events, malformed payloads, and remote fallback cannot appear as successful low-cost work. +Capture fixtures are preconstructed outside measurement: event `benchmark`, identity `benchmark-user-`, and separate caller-owned property dictionaries containing exactly 32 or 4,096 ASCII `x` characters as `payload`, `plan=pro`, and integer `sequence`. Warmup delivery is flushed and verified before receiver/capture state is reset; measured sequences restart at zero. Capture checks require the correct event name, exact payload and plan, sequence-to-identity mapping, and exactly one delivery of each expected sequence. Balanced loss/duplication fails even with matching totals and per-user counts; arrival order is unrestricted. Rejected captures, timeouts that leave undelivered events, malformed payloads, and remote fallback cannot appear as successful low-cost work. ### Capture measurement boundaries `capture()` returns after foreground preparation/enqueue, while a consumer can concurrently prepare its batch. There is no public completion boundary for that background preparation short of flushing. Therefore: -- `capture.enqueue.*` reports foreground elapsed time and **caller-thread CPU**, not process-wide CPU or allocation totals. -- `capture.drain.*` includes `flush(timeout_seconds=None)` and reports the complete delivery path's process CPU and allocations, including requests/serialization. The external worker timeout bounds this wait. Verification does not flush again: unfinished measured work cannot be rescued outside the measurement. Server CPU and server allocations are excluded; elapsed time includes loopback transport. +- `capture.enqueue.*` records `completion_boundary=api_return` and reports foreground elapsed time and **caller-thread CPU**, not process-wide CPU or allocation totals. Its explicit flush and delivery checks occur afterward, outside measurement. +- `capture.drain.steady.*` and `capture.drain.burst.*` record `completion_boundary=completed_delivery`. Steady flushes every 100 captures; burst flushes once after all captures. Parameters record `drain_strategy` and `flush_every`. Both include `flush(timeout_seconds=None)` and reports the complete delivery path's process CPU and allocations, including requests/serialization. The external worker timeout bounds this wait. Verification does not flush again: unfinished measured work cannot be rescued outside the measurement. Server CPU and server allocations are excluded; elapsed time includes loopback transport. - Enqueue-only allocation metrics are explicitly unavailable and direct readers to the drain workload. -The enqueue workload uses 1,000 events, below queue/batch-count limits; partial batches can still be prepared by the consumer. These metrics describe their named paths rather than pretending Python and Node have identical asynchronous lifecycles. +Capture uses public options `gzip=False`, `capture_mode="v0"`, `max_queue_size=100001`, `flush_at=100000`, and one-hour flush/poll intervals. The native consumer and size-driven batching remain active: burst describes the caller flush schedule, not retained queue size or a guaranteed HTTP request count. These metrics describe their named paths rather than pretending Python and Node have identical asynchronous lifecycles. ## Measurement methods Base/head alternate first position between independent repetitions and never benchmark concurrently. The same head-side fixtures and measurement code run against both installations. The runner verifies matching Python, profiler, psutil, OS, and architecture versions. Both workers use a fixed hash seed and a clean Python configuration environment. -Warmup and correctness checks happen outside timing. Timing, allocation profiling, and memory observation are separate passes. Allocation passes use smaller fixed operation counts where tracing every Python allocation would otherwise produce excessive profiler files; the actual count is included in each result and used as its denominator. These counts are identical for base and head. +Warmup and correctness checks happen outside measurement. Warmup uses two refreshes, otherwise the smaller of the operation count and 32 (1,000-flag workloads) or 160 (other workloads); both revisions use the same policy. Timing, allocation profiling, and memory observation are separate passes. Allocation passes use smaller fixed operation counts where tracing every Python allocation would otherwise produce excessive profiler files; the actual count is included in each result and used as its denominator. These counts are identical for base and head. ### Time and CPU @@ -95,23 +98,25 @@ Memray runs with `trace_python_allocators=True` and `FileFormat.ALL_ALLOCATIONS` Profiler capture and aggregation do not contribute to reported timing. Profiles are streamed and removed after aggregation so repeated runs do not retain large binary captures. Raw counters and profiler versions remain in the JSON. -The control suite demonstrates that this configuration records freed small allocations that default Memray tracking does not expose as Python allocator requests. Other controls preserve SDK outputs while adding CPU work and allocation churn; both must be detected. See [Memray's allocator documentation](https://bloomberg.github.io/memray/python_allocators.html) and [API documentation](https://bloomberg.github.io/memray/api.html). +The controls also cover full/smoke inventory, flush schedules, phase reset and input ownership, reordered delivery, balanced duplicates/loss, failed delivery without a rescuing flush, refresh resets, and deterministic memory checkpoints. The control suite demonstrates that this configuration records freed small allocations that default Memray tracking does not expose as Python allocator requests. Other controls preserve SDK outputs while adding CPU work and allocation churn; both must be detected. See [Memray's allocator documentation](https://bloomberg.github.io/memray/python_allocators.html) and [API documentation](https://bloomberg.github.io/memray/api.html). ### Memory - Idle RSS: after imports, definitions, validation, warmup, and settling. -- Observed workload peak RSS: operation checkpoints and end-of-work observation; can miss intervening spikes. +- Observed workload peak RSS: maximum of before-work, operation checkpoints, and end-of-work observations; can miss intervening spikes. Checkpoints follow each 32 completed load evaluations, each refresh, or each 100 captures, including a final partial interval. Steady checkpoints follow scheduled flushes; burst checkpoints precede its final flush. A separate after-work snapshot includes completed delivery. - Lifetime peak RSS: `resource.getrusage`, normalized from platform-specific units. Includes setup and warmup, not just measured work. - Post-settling RSS: after two `gc.collect()` calls separated by a scheduler yield. Python reference counting also frees objects during ordinary execution. -CPython has no uninstrumented total managed-heap byte counter equivalent to Node's `heapUsed`. `post_settling_heap_used` is explicitly unsupported; enabling tracing in the RSS pass would change the measurement. Retained RSS alone does not prove a leak because allocators can keep arenas resident. A public-entry control deliberately retains 8 MiB on each refresh and verifies that post-GC checkpoints expose the growth. +CPython has no uninstrumented total managed-heap byte counter equivalent to Node's `heapUsed`. `post_settling_heap_used` is explicitly unsupported; enabling tracing in the RSS pass would change the measurement. Retained RSS alone does not prove a leak because allocators can keep arenas resident. Natural GC stays enabled, but no forced GC runs during work or checkpoint callbacks. Explicit settling occurs before and after memory measurement; final correctness sweeps follow the post-settling snapshot. A public-entry control deliberately retains 8 MiB on each refresh and verifies that checkpoint RSS exposes the growth. -Each worker has a two-minute timeout and an externally monitored **512 MiB RSS ceiling**, checked approximately every 20 ms. This is a safety bound, not an exact OS-enforced limit: short spikes can exceed it between observations. A ceiling violation is reported as `error` with `rss_limit_exceeded`, not falsely described as a confirmed OS OOM. An explicit Python `MemoryError` is classified as OOM; a generic kill is not. The suite stops after two worker timeouts, after completing the current base/head pair, and preserves partial results for upload. Worker temporary files, including profiler traces, are owned and removed by the coordinator even when a worker is killed. +SDK identity subprocesses have a two-minute deadline. Each worker has a two-minute timeout and an externally monitored **512 MiB RSS ceiling**, checked approximately every 20 ms. This is a safety bound, not an exact OS-enforced limit: short spikes can exceed it between observations. A ceiling violation is reported as `error` with `rss_limit_exceeded`, not falsely described as a confirmed OS OOM. An explicit Python `MemoryError` is classified as OOM; a generic kill is not. The suite stops after two worker timeouts, after completing the current base/head pair, and preserves partial results for upload. Results retain the complete selected workload inventory: unvisited workloads are marked `error` with a not-run reason and no samples, so the failed shard can still be collected and published. Worker temporary files, including profiler traces, are owned and removed by the coordinator even when a worker is killed. ## Results and CI -The pilot JSON format uses `schema_version: 1`, with the same top-level revision, suite, execution, benchmark, metric, sample, and status structure as Node. Measurement IDs, units, method names, and quality are explicit. Python's intentional differences are allocator-call counts, foreground-thread CPU for enqueue, unavailable uninstrumented heap size, and a separate HTTP receiver. Shared schema finalization and rendering are separate work. +The shared JSON format uses `schema_version: 1`, with the same top-level revision, suite, execution, benchmark, metric, sample, and status structure as Node. Measurement IDs, units, method names, and quality are explicit. Python's intentional differences are allocator-call counts, foreground-thread CPU for enqueue, unavailable uninstrumented heap size, and a separate HTTP receiver. The contract, semantic validation, comparisons, and rendering live in [PostHog/.github](https://github.com/PostHog/.github/blob/main/.github/workflows/sdk-benchmark-report.md). + +The workflow runs for every pull request opened, reopened, or updated in `PostHog/posthog-python`. Four independent runner jobs build the merge base and exact head into isolated environments and each run one fixed shard. All six alternating paired repetitions and native measurement passes for a workload stay on one runner, with fresh worker processes and no same-host measurement concurrency. Shard jobs have a ten-minute safety timeout, with an eight-minute measurement step limit to leave time for artifact upload, and do not cancel siblings on failure. Each uploads `base.json`, `head.json`, and `revisions.json` in its own `benchmark-shard-` artifact, including available partial results after failure. -The PR workflow builds the merge base and exact head into isolated environments, then runs this head-side suite against both. It requests read-only permissions, uses pinned actions, requires no service credentials, preserves partial results after worker failures, and uploads JSON artifacts linked from the job summary. +A separate read-only, five-minute collection job downloads the shard artifacts and invokes the SHA-pinned reporting action in `PostHog/.github` with the head-side manifest and matrix result. It validates complete coverage and renders one comparison in the job summary; a failed source job suppresses numeric comparisons. Suite metadata, including the source digest covering `shards.json`, is identical across shards. Each document records its shard as `execution.shard_id`, retaining each runner's environment and SDK build identity. The canonical `python-benchmark-results` artifact contains only `bundle.json` and `revisions.json`; Markdown and comparison JSON are uploaded separately as `python-benchmark-report`. The workflow requests read-only permissions, uses pinned actions, and requires no service credentials. -Correctness failures and broken execution fail the job. Performance differences do not: this producer does not compute regression verdicts. Shared comparisons, noise classification, and PR comments remain separate. Calibrate on the actual Linux runner before introducing performance gates; local macOS results are not a CI noise floor. +Correctness failures and broken execution fail the job. Performance differences do not: this producer does not compute regression verdicts. Comparisons expose descriptive medians and variability rather than calibrated regression verdicts. A separate `workflow_run` caller uses the SHA-pinned reusable workflow in `PostHog/.github` to verify source-run/artifact/PR identities and publish comments without executing PR code with write permissions. This caller must land on the default branch before automatic comments are active, including for fork PRs. Calibrate on the actual Linux runner before introducing performance gates; local macOS results are not a CI noise floor. diff --git a/benchmarks/fixtures.py b/benchmarks/fixtures.py index a5959958..e85a8f57 100644 --- a/benchmarks/fixtures.py +++ b/benchmarks/fixtures.py @@ -1,4 +1,4 @@ -"""Deterministic fixtures matching the Node pilot's flag mix and identities.""" +"""Deterministic fixtures for sdk-benchmarks/v1.""" import hashlib import json @@ -189,7 +189,8 @@ def workloads(smoke=False): "id": f"capture.enqueue.{size}", "mode": "capture", "payload_bytes": size, - "operations": 100 if smoke else 1000, + "operations": 200 if smoke else 1000, + "completion_boundary": "api_return", "concurrency": 1, "passes": ["timing"], } @@ -197,13 +198,19 @@ def workloads(smoke=False): ], *[ { - "id": f"capture.drain.{size}", + "id": f"capture.drain.{strategy}.{size}", "mode": "drain", + "drain_strategy": strategy, + "flush_every": 100 + if strategy == "steady" + else (200 if smoke else 1000), + "completion_boundary": "completed_delivery", "payload_bytes": size, "operations": 200 if smoke else 1000, "concurrency": 1, "passes": ["timing", "allocation", "memory"], } + for strategy in ("steady", "burst") for size in (32, 4096) ], ] diff --git a/benchmarks/identity.py b/benchmarks/identity.py index 6e77152b..263f4d91 100644 --- a/benchmarks/identity.py +++ b/benchmarks/identity.py @@ -13,7 +13,10 @@ def source_digest(directory): str(path.relative_to(directory)): digest(path.read_bytes()) for path in sorted(directory.rglob("*")) if path.is_file() - and path.suffix in (".py", ".txt", ".in") + and ( + path.suffix in (".py", ".txt", ".in") + or path == directory / "shards.json" + ) and "__pycache__" not in path.parts } ) diff --git a/benchmarks/run.py b/benchmarks/run.py index d7fb14f9..f3eca2d4 100644 --- a/benchmarks/run.py +++ b/benchmarks/run.py @@ -4,6 +4,7 @@ import json import math import os +import re import subprocess import sys import tempfile @@ -25,6 +26,47 @@ RSS_LIMIT = 512 * 1024 * 1024 +def load_shards(path=HERE / "shards.json"): + manifest = json.loads(path.read_text()) + if ( + not isinstance(manifest, dict) + or set(manifest) != {"manifest_version", "sdk", "shards"} + or type(manifest["manifest_version"]) is not int + or manifest["manifest_version"] != 1 + or manifest["sdk"] != "posthog-python" + or not isinstance(manifest["shards"], list) + or not 1 <= len(manifest["shards"]) <= 16 + ): + raise ValueError("Invalid shard manifest") + shards = {} + assigned = [] + for shard in manifest["shards"]: + if ( + not isinstance(shard, dict) + or set(shard) != {"id", "workloads"} + or not isinstance(shard["id"], str) + or not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", shard["id"]) + or shard["id"] in shards + or not isinstance(shard["workloads"], list) + or not shard["workloads"] + or not all(isinstance(item, str) for item in shard["workloads"]) + ): + raise ValueError("Invalid or duplicate shard") + shards[shard["id"]] = shard["workloads"] + assigned.extend(shard["workloads"]) + for smoke in (False, True): + inventory = [spec["id"] for spec in workloads(smoke)] + if ( + len(set(assigned)) != len(assigned) + or len(set(inventory)) != len(inventory) + or set(assigned) != set(inventory) + ): + raise ValueError( + "Shard workloads must exactly partition full and smoke inventories" + ) + return shards + + def worker_env(): env = { key: value for key, value in os.environ.items() if not key.startswith("PYTHON") @@ -136,6 +178,7 @@ def document_for(directory, role, pair_id, suite): cwd=cwd, env=worker_env(), text=True, + timeout=TIMEOUT, ) ) execution.update( @@ -297,19 +340,32 @@ def main(args=None): parser.add_argument("--output", type=Path, required=True) parser.add_argument("--repetitions", type=int, default=6, choices=range(1, 31)) parser.add_argument("--smoke", action="store_true") - parser.add_argument("--filter", default="") + selection = parser.add_mutually_exclusive_group() + selection.add_argument("--filter", default="") + selection.add_argument("--shard") options = parser.parse_args(args) + try: + shards = load_shards() + except ValueError as error: + parser.error(str(error)) + if options.shard is not None and options.shard not in shards: + parser.error(f"Unknown shard: {options.shard}") specs = [ spec for spec in workloads(options.smoke) - if spec["id"].startswith(options.filter) + if ( + spec["id"] in shards[options.shard] + if options.shard is not None + else spec["id"].startswith(options.filter) + ) ] if not specs: parser.error("No matching workloads") pair_id = str(uuid.uuid4()) suite = { "id": "python-sdk-performance", - "version": 1, + "version": 2, + "specification": "sdk-benchmarks/v1", "source_commit": git(HERE, "rev-parse", "HEAD"), "source_digest": source_digest(HERE), "profile": "smoke" if options.smoke else "full", @@ -320,13 +376,15 @@ def main(args=None): "scope": "PYMALLOC_MALLOC/CALLOC/REALLOC requests; all worker threads", }, "transport": "loopback HTTP receiver in coordinator process; uncompressed v0", + "warmup": "two refreshes; otherwise min(operations, 32 for 1000 flags or 160); verified phase reset before measurement", "settling": "two gc.collect calls separated by sleep(0)", "memory_sampling": "operation checkpoints; lifetime high-water mark includes setup and warmup", "capture": { "flush_at": 100000, "flush_interval_seconds": 3600, "consumer_threads": 1, - "drain_every": 100, + "max_queue_size": 100001, + "compression": "none", }, "flag_poll_interval_seconds": 3600, } @@ -335,9 +393,35 @@ def main(args=None): role: document_for(directory, role, pair_id, suite) for role, directory in directories.items() } + if options.shard is not None: + for document in documents.values(): + document["execution"]["shard_id"] = options.shard for key in ("runtime_version", "measurement_tools", "os", "arch"): if documents["base"]["execution"][key] != documents["head"]["execution"][key]: raise ValueError(f"Base/head {key} differ") + for document in documents.values(): + document["benchmarks"] = [ + { + "id": spec["id"], + "parameters": spec, + "fixture_digest": digest( + { + "spec": spec, + "definitions": ( + definitions(spec["flag_count"]) + if "flag_count" in spec + else None + ), + "users": USERS, + } + ), + "status": "error", + "reason": "Workload was not run: suite stopped before reaching it", + "metrics": [], + "runs": [], + } + for spec in specs + ] options.output.mkdir(parents=True, exist_ok=True) def save(): @@ -348,26 +432,12 @@ def save(): failures = 0 timeouts = 0 - for spec in specs: + for index, spec in enumerate(specs): records = {} for role in directories: - records[role] = { - "id": spec["id"], - "parameters": spec, - "fixture_digest": digest( - { - "spec": spec, - "definitions": definitions(spec["flag_count"]) - if "flag_count" in spec - else None, - "users": USERS, - } - ), - "status": "ok", - "metrics": [], - "runs": [], - } - documents[role]["benchmarks"].append(records[role]) + records[role] = documents[role]["benchmarks"][index] + records[role]["status"] = "ok" + del records[role]["reason"] for repetition in range(options.repetitions): for measurement_pass in spec["passes"]: for role in ["head", "base"] if repetition % 2 else ["base", "head"]: diff --git a/benchmarks/server.py b/benchmarks/server.py index b513791b..b119c2ff 100644 --- a/benchmarks/server.py +++ b/benchmarks/server.py @@ -20,11 +20,15 @@ def __init__(self, spec, batch_delay=0): self.lock = threading.Lock() self.payload = "x" * spec.get("payload_bytes", 0) self.body = json.dumps(definitions(spec.get("flag_count", 0))).encode() + self.seen = bytearray(spec["operations"]) + self.expected_sequences = 0 self.state: dict[str, Any] = { "definitions": 0, "delivered": 0, "requests": 0, "invalid_events": 0, + "unique_sequences": 0, + "missing_sequences": 0, "delivered_by_user": {user["distinct_id"]: 0 for user in USERS}, "unexpected": [], } @@ -65,7 +69,29 @@ def do_POST(self): if not 0 < size <= 20 * 1024 * 1024: raise ValueError("Invalid body length") body = json.loads(self.rfile.read(size)) - if path == "/__benchmark/definitions": + if path == "/__benchmark/reset": + operations = body["operations"] + if ( + type(operations) is not int + or not 0 <= operations <= spec["operations"] + ): + raise ValueError("Invalid phase operations") + owner.seen[:] = bytes(len(owner.seen)) + owner.expected_sequences = operations + owner.state.update( + definitions=0, + delivered=0, + requests=0, + invalid_events=0, + unique_sequences=0, + missing_sequences=operations, + delivered_by_user={ + user["distinct_id"]: 0 for user in USERS + }, + unexpected=[], + ) + self.reply(200, b"{}") + elif path == "/__benchmark/definitions": generation = body["generation"] if ( not isinstance(generation, int) @@ -87,15 +113,29 @@ def do_POST(self): for event in batch: properties = event.get("properties", {}) user = event.get("distinct_id") + sequence = properties.get("sequence") + valid_sequence = ( + type(sequence) is int + and 0 <= sequence < owner.expected_sequences + ) if ( event.get("event") != "benchmark" or user not in owner.state["delivered_by_user"] or properties.get("payload") != owner.payload or properties.get("plan") != "pro" + or not valid_sequence + or user + != USERS[sequence % len(USERS)]["distinct_id"] ): owner.state["invalid_events"] += 1 else: owner.state["delivered_by_user"][user] += 1 + if owner.seen[sequence]: + owner.state["invalid_events"] += 1 + else: + owner.seen[sequence] = 1 + owner.state["unique_sequences"] += 1 + owner.state["missing_sequences"] -= 1 self.reply(200, b'{"status":1}') else: self.unexpected(path) diff --git a/benchmarks/shards.json b/benchmarks/shards.json new file mode 100644 index 00000000..faa3085a --- /dev/null +++ b/benchmarks/shards.json @@ -0,0 +1,39 @@ +{ + "manifest_version": 1, + "sdk": "posthog-python", + "shards": [ + { + "id": "flags-single", + "workloads": [ + "flags.single.boolean", + "flags.single.rollout", + "flags.single.variant", + "flags.single.property", + "flags.single.regex", + "flags.single.cohort", + "flags.single.early", + "flags.single.late", + "flags.single.no-match" + ] + }, + { + "id": "flags-scaling", + "workloads": ["flags.all.10", "flags.all.100", "flags.all.1000"] + }, + { + "id": "flags-memory", + "workloads": ["flags.load.1000", "flags.refresh.1000"] + }, + { + "id": "capture", + "workloads": [ + "capture.enqueue.32", + "capture.enqueue.4096", + "capture.drain.steady.32", + "capture.drain.steady.4096", + "capture.drain.burst.32", + "capture.drain.burst.4096" + ] + } + ] +} diff --git a/benchmarks/test_benchmarks.py b/benchmarks/test_benchmarks.py index b5dd1998..af51f411 100644 --- a/benchmarks/test_benchmarks.py +++ b/benchmarks/test_benchmarks.py @@ -1,4 +1,6 @@ +import copy import gc +import io import json import subprocess import sys @@ -22,8 +24,9 @@ time_operations, ) from prepare import prepare -from run import append_sample, main, run_worker +from run import append_sample, load_shards, main, run_worker from server import BenchmarkServer +from worker import execute from workload import Workload, control from posthog import Posthog @@ -33,6 +36,449 @@ class BenchmarksTest(unittest.TestCase): + def test_specification_inventory_and_parameters(self): + single_cases = ( + "boolean", + "rollout", + "variant", + "property", + "regex", + "cohort", + "early", + "late", + "no-match", + ) + for smoke in (False, True): + expected = { + **{ + f"flags.single.{case}": ( + 160 if smoke else 16000, + ["timing", "allocation"], + ) + for case in single_cases + }, + **{ + f"flags.all.{count}": ( + 16 if smoke else operations, + ["timing", "allocation"], + ) + for count, operations in ((10, 1600), (100, 160), (1000, 160)) + }, + "flags.load.1000": (32 if smoke else 1600, ["memory"]), + "flags.refresh.1000": (4 if smoke else 40, ["memory"]), + **{ + f"capture.enqueue.{size}": (200 if smoke else 1000, ["timing"]) + for size in (32, 4096) + }, + **{ + f"capture.drain.{strategy}.{size}": ( + 200 if smoke else 1000, + ["timing", "allocation", "memory"], + ) + for strategy in ("steady", "burst") + for size in (32, 4096) + }, + } + specs = workloads(smoke) + self.assertEqual(len(specs), 20) + self.assertEqual( + {s["id"]: (s["operations"], s["passes"]) for s in specs}, expected + ) + for spec in specs: + self.assertEqual( + spec["concurrency"], 8 if spec["id"] == "flags.load.1000" else 1 + ) + if spec["mode"] == "single": + self.assertEqual(spec["flag_count"], 10) + self.assertEqual( + spec["flag_index"], + single_cases.index(spec["id"].split(".")[-1]), + ) + self.assertEqual( + spec["allocation_operations"], 160 if smoke else 2000 + ) + elif "flag_count" in spec: + self.assertEqual(spec["flag_count"], int(spec["id"].split(".")[-1])) + if "allocation_operations" in spec: + self.assertEqual( + spec["allocation_operations"], + 16 if smoke else max(16, 1600 // spec["flag_count"]), + ) + else: + self.assertEqual( + spec["payload_bytes"], int(spec["id"].split(".")[-1]) + ) + self.assertEqual( + spec["completion_boundary"], + "api_return" + if spec["mode"] == "capture" + else "completed_delivery", + ) + if spec["mode"] == "drain": + self.assertEqual( + spec["drain_strategy"], spec["id"].split(".")[2] + ) + self.assertEqual( + spec["flush_every"], + 100 + if spec["drain_strategy"] == "steady" + else spec["operations"], + ) + + def test_shard_manifest_exactly_partitions_full_and_smoke(self): + shards = load_shards() + self.assertEqual( + list(shards), ["flags-single", "flags-scaling", "flags-memory", "capture"] + ) + assigned = [item for items in shards.values() for item in items] + for smoke in (False, True): + self.assertEqual(assigned, [spec["id"] for spec in workloads(smoke)]) + self.assertEqual([len(items) for items in shards.values()], [9, 3, 2, 6]) + with ( + patch("run.workloads", side_effect=[workloads(), workloads(True)[:-1]]), + self.assertRaisesRegex(ValueError, "partition"), + ): + load_shards() + + def test_shard_manifest_rejects_invalid_partitions_and_identifiers(self): + original = json.loads((ROOT / "benchmarks/shards.json").read_text()) + for case in ( + "version", + "sdk", + "empty", + "duplicate-shard", + "unsafe-id", + "empty-workloads", + "duplicate-within", + "duplicate-across", + "missing", + "unknown", + "extra-field", + "invalid-workload", + ): + manifest = copy.deepcopy(original) + shards = manifest["shards"] + if case == "version": + manifest["manifest_version"] = True + elif case == "sdk": + manifest["sdk"] = "posthog-node" + elif case == "empty": + manifest["shards"] = [] + elif case == "duplicate-shard": + shards[1]["id"] = shards[0]["id"] + elif case == "unsafe-id": + shards[0]["id"] = "../flags-single" + elif case == "empty-workloads": + shards[0]["workloads"] = [] + elif case == "duplicate-within": + shards[0]["workloads"].append(shards[0]["workloads"][0]) + elif case == "duplicate-across": + shards[1]["workloads"].append(shards[0]["workloads"][0]) + elif case == "missing": + shards[0]["workloads"].pop() + elif case == "unknown": + shards[0]["workloads"][0] = "flags.unknown" + elif case == "extra-field": + manifest["extra"] = 1 + elif case == "invalid-workload": + shards[0]["workloads"][0] = [] + with self.subTest(case=case), tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "shards.json" + path.write_text(json.dumps(manifest)) + with self.assertRaises(ValueError): + load_shards(path) + + def test_shard_and_filter_cli_errors_precede_execution(self): + for selection in ( + ["--shard", "unknown"], + ["--shard", "capture", "--filter", "capture."], + ["--shard", "capture", "--filter", ""], + ["--filter", "unknown"], + ): + with ( + self.subTest(selection=selection), + patch("run.document_for") as describe, + patch("sys.stderr", new_callable=io.StringIO), + self.assertRaises(SystemExit) as error, + ): + main( + [ + "--base", + str(ROOT), + "--head", + str(ROOT), + "--output", + "/unused", + *selection, + ] + ) + self.assertEqual(error.exception.code, 2) + describe.assert_not_called() + + def test_shards_preserve_suite_and_complete_alternating_repetitions(self): + suites = [] + pair_ids = [] + for shard_id, selected in load_shards().items(): + with ( + self.subTest(shard=shard_id), + tempfile.TemporaryDirectory() as directory, + patch("run.run_worker", return_value={"status": "ok"}) as worker, + patch("run.append_sample") as append, + patch("sys.stderr", new_callable=io.StringIO), + ): + self.assertEqual( + main( + [ + "--base", + str(ROOT), + "--head", + str(ROOT), + "--output", + directory, + "--smoke", + "--shard", + shard_id, + ] + ), + 0, + ) + base = json.loads((Path(directory) / "base.json").read_text()) + head = json.loads((Path(directory) / "head.json").read_text()) + self.assertEqual(base["suite"], head["suite"]) + suites.append(head["suite"]) + self.assertEqual(head["suite"]["repetitions"], 6) + self.assertEqual( + base["execution"]["pair_id"], head["execution"]["pair_id"] + ) + pair_ids.append(head["execution"]["pair_id"]) + for document in (base, head): + self.assertEqual(document["execution"]["shard_id"], shard_id) + self.assertEqual( + [b["id"] for b in document["benchmarks"]], selected + ) + specs = [spec for spec in workloads(True) if spec["id"] in selected] + self.assertEqual( + [(call.args[1], call.args[2]) for call in worker.call_args_list], + [ + (spec, measurement_pass) + for spec in specs + for _ in range(6) + for measurement_pass in spec["passes"] + for _ in range(2) + ], + ) + expected = [ + (spec["id"], repetition, role) + for spec in specs + for repetition in range(6) + for _ in spec["passes"] + for role in ( + ["head", "base"] if repetition % 2 else ["base", "head"] + ) + ] + # Record identity distinguishes the two sides even for same-revision runs. + records = { + id(call.args[0]): call.args[0] for call in append.call_args_list + } + roles: dict[int, str] = {} + for spec in specs: + matching = [ + key + for key, record in records.items() + if record["id"] == spec["id"] + ] + roles.update(zip(matching, ("base", "head"))) + self.assertEqual( + [ + (call.args[0]["id"], call.args[2], roles[id(call.args[0])]) + for call in append.call_args_list + ], + expected, + ) + self.assertTrue(all(suite == suites[0] for suite in suites)) + self.assertEqual(len(set(pair_ids)), 4) + + def test_capture_phase_reset_flush_schedules_and_fixture_ownership(self): + for strategy in ("steady", "burst"): + calls: list[int] = [] + submitted: list[dict[str, Any]] = [] + + class Observed(Posthog): + def capture(self, event, _submitted=submitted, **kwargs): + _submitted.append(kwargs["properties"]) + return super().capture(event, **kwargs) + + def flush(self, timeout_seconds=10, _calls=calls, _submitted=submitted): + _calls.append(len(_submitted)) + return super().flush(timeout_seconds=timeout_seconds) + + spec = next( + s for s in workloads(True) if s["id"] == f"capture.drain.{strategy}.32" + ) + with self.subTest(strategy=strategy), BenchmarkServer(spec) as server: + workload = Workload(spec, server.url, client_class=Observed) + try: + workload.prepare_phase(160) + workload.run(160) + workload.verify() + old_properties = list(submitted) + workload.prepare_phase(spec["operations"]) + self.assertEqual(workload.state()["delivered"], 0) + calls.clear() + submitted.clear() + checkpoints: list[tuple[int, list[int]]] = [] + with patch( + "gc.collect", + side_effect=AssertionError("GC inside measured work"), + ): + workload.run( + checkpoint=lambda n, points=checkpoints, flushes=calls: ( + points.append((n, list(flushes))) + ) + ) + self.assertEqual( + calls, [100, 200] if strategy == "steady" else [200] + ) + self.assertEqual( + checkpoints, + [(100, [100]), (200, [100, 200])] + if strategy == "steady" + else [(100, []), (200, [])], + ) + count = len(calls) + state = workload.verify() + self.assertEqual(len(calls), count) + self.assertEqual(state["unique_sequences"], 200) + self.assertEqual(state["missing_sequences"], 0) + self.assertEqual( + [p["sequence"] for p in submitted], list(range(200)) + ) + self.assertEqual( + len({id(p) for p in submitted + old_properties}), 360 + ) + self.assertEqual( + [p["sequence"] for p in old_properties], list(range(160)) + ) + finally: + workload.close() + + def test_balanced_duplicate_loss_fails_and_delivery_order_is_not_required(self): + spec = next(s for s in workloads(True) if s["id"] == "capture.drain.burst.32") + for duplicate in (False, True): + with self.subTest(duplicate=duplicate), BenchmarkServer(spec) as server: + workload = Workload(spec, server.url) + try: + events = workload.events + if duplicate: + # Replace the final 16 events with the preceding 16: total and + # per-identity counts stay identical, but sequences do not. + events = events[:-16] + events[-32:-16] + control(server.url, "/batch", {"batch": list(reversed(events))}) + workload.captured = spec["operations"] + for event in workload.events: + workload.captured_by_user[event["distinct_id"]] += 1 + self.assertEqual( + workload.state()["delivered_by_user"], workload.captured_by_user + ) + if duplicate: + with self.assertRaises(AssertionError): + workload.verify() + self.assertEqual(workload.state()["missing_sequences"], 16) + else: + workload.verify() + finally: + workload.close() + + def test_incomplete_delivery_is_not_rescued_by_verification(self): + calls = [] + + class Incomplete(Posthog): + def flush(self, timeout_seconds=10): + calls.append(timeout_seconds) + + spec = next(s for s in workloads(True) if s["id"] == "capture.drain.burst.32") + with BenchmarkServer(spec) as server: + workload = Workload(spec, server.url, client_class=Incomplete) + try: + workload.run() + with self.assertRaises(AssertionError): + workload.verify() + self.assertEqual(calls, [None]) + finally: + Posthog.flush(workload.client, timeout_seconds=None) + workload.close() + + def test_refresh_generation_reset_and_full_sweeps_outside_measurement(self): + spec = next(s for s in workloads(True) if s["mode"] == "refresh") + with BenchmarkServer(spec) as server: + workload = Workload(spec, server.url) + try: + workload.run(2) + workload.verify() + workload.prepare_phase(spec["operations"]) + self.assertEqual(workload.generation, 0) + for generation in range(1, 5): + workload.run(1) + self.assertEqual(workload.generation, generation) + workload.verify() + workload.prepare_phase(spec["operations"]) + checks = workload.checks + checkpoints: list[int] = [] + with patch( + "gc.collect", side_effect=AssertionError("GC inside measured work") + ): + workload.run(checkpoint=checkpoints.append) + self.assertEqual(workload.checks, checks) + self.assertEqual(checkpoints, [1, 2, 3, 4]) + workload.verify() + self.assertEqual(workload.checks, checks + 16) + finally: + workload.close() + + def test_memory_checkpoint_positions_and_settling_order(self): + for workload_id, operations, positions in ( + ("flags.load.1000", 70, [32, 64, 70]), + ("flags.refresh.1000", 4, [1, 2, 3, 4]), + ("capture.drain.steady.32", 250, [100, 200, 250]), + ("capture.drain.burst.32", 250, [100, 200, 250]), + ): + spec = { + **next(s for s in workloads(True) if s["id"] == workload_id), + "operations": operations, + "pass": "memory", + } + events: list[str] = [] + + class Observed(Workload): + def run(self, *args, _events=events, **kwargs): + _events.append("run") + with patch( + "gc.collect", + side_effect=AssertionError("GC inside measured work"), + ): + return super().run(*args, **kwargs) + + def verify(self, _events=events): + _events.append("verify") + return super().verify() + + with ( + self.subTest(workload=workload_id), + BenchmarkServer(spec) as server, + patch("worker.Workload", Observed), + patch( + "worker.settle", side_effect=lambda log=events: log.append("settle") + ), + patch( + "worker.memory", + side_effect=lambda log=events: log.append("memory") or memory(), + ), + ): + raw = execute(spec, server.url) + self.assertEqual([p["operations"] for p in raw["checkpoints"]], positions) + self.assertEqual(events[-4:], ["memory", "settle", "memory", "verify"]) + self.assertEqual(events.count("settle"), 2) + def test_deterministic_and_changing_fixtures(self): self.assertEqual(digest(definitions(1000)), digest(definitions(1000))) self.assertNotEqual(digest(definitions(1000)), digest(definitions(1000, 1))) @@ -102,7 +548,7 @@ def flush(self, timeout_seconds=10): ) spec = { - **next(s for s in workloads(True) if s["id"] == "capture.drain.32"), + **next(s for s in workloads(True) if s["id"] == "capture.drain.steady.32"), "operations": 100, } with BenchmarkServer(spec, batch_delay=0.05) as server: @@ -211,7 +657,6 @@ def load_feature_flags(self): checkpoints = [] def checkpoint(completed): - settle() checkpoints.append((completed, memory()["rss"])) workload.run(checkpoint=checkpoint) @@ -295,32 +740,52 @@ def test_killed_worker_temporary_files_are_removed_by_the_coordinator(self): self.assertTrue(profile.parent.name.startswith("posthog-benchmark-worker-")) self.assertFalse(profile.parent.exists()) - def test_repeated_timeouts_stop_the_suite_with_partial_results(self): - with ( - tempfile.TemporaryDirectory() as directory, - patch( - "run.run_worker", return_value={"status": "timeout", "pass": "timing"} - ) as worker, - ): - code = main( - [ - "--base", - str(ROOT), - "--head", - str(ROOT), - "--output", - directory, - "--smoke", - ] - ) - self.assertEqual(code, 1) - self.assertEqual(worker.call_count, 2) - for role in ("base", "head"): - document = json.loads((Path(directory) / f"{role}.json").read_text()) - self.assertEqual(len(document["benchmarks"]), 1) - self.assertEqual(document["benchmarks"][0]["status"], "timeout") - self.assertEqual(document["benchmarks"][0]["metrics"], []) - self.assertEqual(len(document["benchmarks"][0]["runs"]), 1) + def test_repeated_timeouts_preserve_the_complete_inventory(self): + for shard in (None, *load_shards()): + with ( + self.subTest(shard=shard), + tempfile.TemporaryDirectory() as directory, + patch( + "run.run_worker", + side_effect=lambda python, spec, measurement_pass: { + "status": "timeout", + "pass": measurement_pass, + }, + ) as worker, + ): + code = main( + [ + "--base", + str(ROOT), + "--head", + str(ROOT), + "--output", + directory, + "--smoke", + *(["--shard", shard] if shard else []), + ] + ) + self.assertEqual(code, 1) + self.assertEqual(worker.call_count, 2) + expected = ( + load_shards()[shard] + if shard + else [spec["id"] for spec in workloads(True)] + ) + for role in ("base", "head"): + document = json.loads( + (Path(directory) / f"{role}.json").read_text() + ) + records = document["benchmarks"] + self.assertEqual([record["id"] for record in records], expected) + self.assertEqual(records[0]["status"], "timeout") + self.assertEqual(records[0]["metrics"], []) + self.assertEqual(len(records[0]["runs"]), 1) + for record in records[1:]: + self.assertEqual(record["status"], "error") + self.assertIn("not run", record["reason"]) + self.assertEqual(record["metrics"], []) + self.assertEqual(record["runs"], []) def test_oom_classification_requires_evidence(self): self.assertEqual(classify_exit(stderr="Killed"), "error") @@ -336,6 +801,11 @@ def test_package_identity_and_source_fingerprints(self): before = source_digest(directory) path.write_text("VALUE = 2") self.assertNotEqual(source_digest(directory), before) + manifest = Path(directory) / "shards.json" + manifest.write_text('{"manifest_version": 1}') + before = source_digest(directory) + manifest.write_text('{"manifest_version": 2}') + self.assertNotEqual(source_digest(directory), before) def test_paired_documents(self): with tempfile.TemporaryDirectory() as directory: @@ -367,6 +837,8 @@ def test_paired_documents(self): base["benchmarks"][0]["fixture_digest"], head["benchmarks"][0]["fixture_digest"], ) + self.assertEqual(base["suite"]["version"], 2) + self.assertEqual(base["suite"]["specification"], "sdk-benchmarks/v1") self.assertEqual(base["revision"]["role"], "base") self.assertEqual(head["revision"]["role"], "head") for metric in head["benchmarks"][0]["metrics"]: diff --git a/benchmarks/worker.py b/benchmarks/worker.py index e8b80f23..eb3f762a 100644 --- a/benchmarks/worker.py +++ b/benchmarks/worker.py @@ -30,14 +30,16 @@ def execute(spec, host): if spec["mode"] == "refresh" else min(spec["operations"], 32 if spec.get("flag_count") == 1000 else 160) ) - workload.run(warmup) - workload.verify() - settle() operations = ( spec.get("allocation_operations", spec["operations"]) if spec["pass"] == "allocation" else spec["operations"] ) + workload.prepare_phase(warmup) + workload.run(warmup) + workload.verify() + workload.prepare_phase(operations) + settle() result = { "pass": spec["pass"], "operations": operations, @@ -61,12 +63,9 @@ def execute(spec, host): def checkpoint(completed): nonlocal observed - observed = max(observed, memory()["rss"]) - if spec["mode"] == "refresh" and ( - completed % 5 == 0 or completed == spec["operations"] - ): - settle() - result["checkpoints"].append({"operations": completed, **memory()}) + snapshot = memory() + observed = max(observed, snapshot["rss"]) + result["checkpoints"].append({"operations": completed, **snapshot}) start = time.perf_counter_ns() workload.run(checkpoint=checkpoint) diff --git a/benchmarks/workload.py b/benchmarks/workload.py index 069824c7..570cde32 100644 --- a/benchmarks/workload.py +++ b/benchmarks/workload.py @@ -49,26 +49,41 @@ def __init__(self, spec, host, client_class=None): if spec["concurrency"] > 1 else None ) - self.events = [ - { - "event": "benchmark", - "distinct_id": user["distinct_id"], - "properties": { - "payload": "x" * spec.get("payload_bytes", 0), - "plan": "pro", - }, - } - for user in USERS - ] try: if self.flags: self.check_flags() assert self.state()["definitions"] > 0 - self.verify() + self.verify() + else: + self.prepare_phase(spec["operations"]) except BaseException: self.close() raise + def prepare_phase(self, operations): + """Reset only outside measurement, after the preceding phase is verified.""" + if self.flags: + self.generation = 0 + control(self.host, "/__benchmark/definitions", {"generation": 0}) + self.client.load_feature_flags() + self.check_flags() + else: + control(self.host, "/__benchmark/reset", {"operations": operations}) + self.captured = 0 + self.captured_by_user = {user["distinct_id"]: 0 for user in USERS} + self.events = [ + { + "event": "benchmark", + "distinct_id": USERS[i % len(USERS)]["distinct_id"], + "properties": { + "payload": "x" * self.spec["payload_bytes"], + "plan": "pro", + "sequence": i, + }, + } + for i in range(operations) + ] + def state(self): return control(self.host, "/__benchmark/state") @@ -115,22 +130,30 @@ def run(self, operations=None, checkpoint=None): before = self.state()["definitions"] self.client.load_feature_flags() assert self.state()["definitions"] == before + 1 - self.check_flags() if checkpoint: checkpoint(i + 1) elif mode in ("capture", "drain"): for i in range(operations): - event = self.events[i % len(self.events)] + event = self.events[i] assert self.client.capture(**event) is not None, ( "Capture rejected an event" ) self.captured += 1 self.captured_by_user[USERS[i % len(USERS)]["distinct_id"]] += 1 - if (i + 1) % 100 == 0 or i + 1 == operations: - if mode == "drain": - self.client.flush(timeout_seconds=None) - if checkpoint: - checkpoint(i + 1) + completed = i + 1 + if ( + mode == "drain" + and self.spec["drain_strategy"] == "steady" + and ( + completed % self.spec["flush_every"] == 0 + or completed == operations + ) + ): + self.client.flush(timeout_seconds=None) + if checkpoint and (completed % 100 == 0 or completed == operations): + checkpoint(completed) + if mode == "drain" and self.spec["drain_strategy"] == "burst": + self.client.flush(timeout_seconds=None) else: for start in range(0, operations, self.spec["concurrency"]): if self.pool: @@ -145,8 +168,9 @@ def run(self, operations=None, checkpoint=None): future.result() else: self.evaluate(start) - if checkpoint and start % 32 == 0: - checkpoint(min(start + self.spec["concurrency"], operations)) + completed = min(start + self.spec["concurrency"], operations) + if checkpoint and (completed % 32 == 0 or completed == operations): + checkpoint(completed) def verify(self): if self.flags: @@ -159,7 +183,11 @@ def verify(self): assert state["delivered"] == self.captured, ( "Delivered counts must match captures" ) - assert state["invalid_events"] == 0, "Capture payload or identity changed" + assert state["invalid_events"] == 0, ( + "Capture payload, identity or sequence changed" + ) + assert state["unique_sequences"] == self.captured + assert state["missing_sequences"] == 0, "Missing capture sequences" assert state["delivered_by_user"] == self.captured_by_user, ( "Capture identity distribution changed" )