diff --git a/.github/workflows/benchmark-report.yml b/.github/workflows/benchmark-report.yml new file mode 100644 index 0000000..6de61dc --- /dev/null +++ b/.github/workflows/benchmark-report.yml @@ -0,0 +1,21 @@ +name: Ruby SDK benchmark report + +on: + workflow_run: + workflows: ['Ruby SDK benchmarks'] + types: [completed] + +permissions: {} + +jobs: + report: + permissions: + contents: read + actions: read + pull-requests: write + uses: PostHog/.github/.github/workflows/sdk-benchmark-report.yml@2a6111d08d784c0ca84f7f97f32a13fdd6d31449 # SDK benchmark workflow (PostHog/.github PR #84) + with: + workflow-path: .github/workflows/benchmarks.yml + artifact-name: ruby-benchmark-results + report-key: posthog-ruby + sdk: posthog-ruby diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml new file mode 100644 index 0000000..061dedf --- /dev/null +++ b/.github/workflows/benchmarks.yml @@ -0,0 +1,149 @@ +name: Ruby SDK benchmarks + +on: [pull_request] + +permissions: {} + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + benchmark: + runs-on: ubuntu-24.04 + timeout-minutes: 12 + env: + BUNDLE_PATH: ${{ github.workspace }}/vendor/bundle + strategy: + fail-fast: false + matrix: + shard: [flags-single, flags-scaling, flags-memory, capture] + permissions: + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha }} + fetch-depth: 0 + persist-credentials: false + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.12.12' + + - uses: ruby/setup-ruby@b1da4809b005e0178fe08a1d57317bffc00b7b4c # v1.332.0 + with: + ruby-version: '3.4.7' + bundler: '4.0.13' + bundler-cache: true + + - name: Prepare pinned baseline + env: + TARGET_SHA: ${{ github.event.pull_request.base.sha }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + set -euo pipefail + export BASE_WORKTREE="$RUNNER_TEMP/ruby-benchmark-base" + export RESULTS="$RUNNER_TEMP/ruby-benchmark-results" + printf 'BASE_WORKTREE=%s\nRESULTS=%s\n' "$BASE_WORKTREE" "$RESULTS" >> "$GITHUB_ENV" + BASE_SHA=$(git merge-base "$TARGET_SHA" "$HEAD_SHA") + git worktree add --detach "$BASE_WORKTREE" "$BASE_SHA" + mkdir -p "$RESULTS" + python -c 'import json, os, pathlib, sys; pathlib.Path(os.environ["RESULTS"], "revisions.json").write_text(json.dumps({"target":os.environ["TARGET_SHA"], "head":os.environ["HEAD_SHA"], "base":sys.argv[1]}, indent=2))' "$BASE_SHA" + + - name: Build and install isolated gems + run: | + python -B benchmarks/prepare.py . "$RUNNER_TEMP/ruby-benchmark-build/head" + python -B benchmarks/prepare.py "$BASE_WORKTREE" "$RUNNER_TEMP/ruby-benchmark-build/base" + + - name: Test measurement controls and fixtures + run: | + python -B benchmarks/test_benchmarks.py + python -B benchmarks/check_native.py "$RUNNER_TEMP/ruby-benchmark-build/head" + + - name: Measure paired base and head + id: measure + timeout-minutes: 10 + run: | + python -B benchmarks/run.py \ + --base "$BASE_WORKTREE" \ + --head "$GITHUB_WORKSPACE" \ + --build-directory "$RUNNER_TEMP/ruby-benchmark-build" \ + --output "$RESULTS" \ + --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 }}/ruby-benchmark-results/base.json + ${{ runner.temp }}/ruby-benchmark-results/head.json + ${{ runner.temp }}/ruby-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 }}/ruby-benchmark-shards + merge-multiple: false + + - name: Collect and render benchmark comparison + uses: PostHog/.github/.github/actions/benchmark-report@04f8421ec4daa7f4066eaeb4c53621ba79ddc3ca # SDK benchmark reporting (PostHog/.github PR #84) + with: + mode: collect + source-status: ${{ needs.benchmark.result }} + shard-manifest: benchmarks/shards.json + shard-directory: ${{ runner.temp }}/ruby-benchmark-shards + output-directory: ${{ runner.temp }}/ruby-benchmark-report + + - name: Upload rendered comparison + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: ruby-benchmark-report + path: | + ${{ runner.temp }}/ruby-benchmark-report/report.md + ${{ runner.temp }}/ruby-benchmark-report/comparison.json + if-no-files-found: ignore + retention-days: 14 + + - name: Upload raw benchmark results + if: always() + id: results + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: ruby-benchmark-results + path: | + ${{ runner.temp }}/ruby-benchmark-report/bundle.json + ${{ runner.temp }}/ruby-benchmark-report/revisions.json + if-no-files-found: warn + retention-days: 14 + + - name: Link results + if: always() + env: + ARTIFACT_URL: ${{ steps.results.outputs.artifact-url }} + run: | + printf '## Ruby SDK benchmark data\n\n' >> "$GITHUB_STEP_SUMMARY" + if [[ -n "$ARTIFACT_URL" ]]; then + printf '[Download paired measurements and execution metadata](%s)\n\n' "$ARTIFACT_URL" >> "$GITHUB_STEP_SUMMARY" + fi + printf 'Measurements are informational. Check workload statuses before comparing values.\n' >> "$GITHUB_STEP_SUMMARY" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 2679fc6..84f3191 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -75,6 +75,37 @@ This section is for external contributors. PostHog maintainers (members of the P A diff in `public_api_snapshot.txt` (see "CI-aligned checks" above) means your change touches public API. +## SDK performance benchmarks + +Every PR runs the 20-workload [`sdk-benchmarks/v1` specification](https://github.com/PostHog/.github/blob/04f8421ec4daa7f4066eaeb4c53621ba79ddc3ca/.github/workflows/sdk-benchmark-specification.md) against its merge base and exact head. The same head-side fixtures run against separately built, installed gems. Four independent workload shards keep all six alternating base/head pairs on one runner, with a fresh Ruby process for each workload, measurement pass and repetition. No performance percentage is a CI gate; build, correctness and execution errors remain failures. + +The benchmark preparation step installs each checkout's frozen Bundler dependencies (with the repository's RubyGems cooldown), builds its SDK gem, and installs that gem into a fresh external directory. Worker processes use those installed gems, not source imports: + +```bash +python3 -B benchmarks/prepare.py /path/to/base /path/to/build/base +python3 -B benchmarks/prepare.py "$PWD" /path/to/build/head +python3 -B benchmarks/test_benchmarks.py +python3 -B benchmarks/check_native.py /path/to/build/head +python3 -B benchmarks/run.py \ + --base /path/to/base --head "$PWD" --build-directory /path/to/build \ + --output /path/to/results --smoke --repetitions 1 +``` + +Omit `--smoke` and use `--repetitions 6` for full measurements. `--shard` selects `flags-single`, `flags-scaling`, `flags-memory` or `capture`; `--filter` selects a workload prefix for debugging and cannot be combined with a shard. Smoke checks retain the complete inventory with smaller operation counts. Full flags allocation passes use the documented smaller counts recorded in `allocation_operations`. + +Measurement scope: + +- Elapsed time uses `Process.clock_gettime(CLOCK_MONOTONIC)`. CPU uses process time, except enqueue-only workloads use caller-thread time because background serialization has no public enqueue completion signal. +- Allocation passes count Ruby objects with cumulative `GC.stat(:total_allocated_objects)`, including objects later collected and allocations on SDK background threads. Native malloc is excluded. Cumulative allocated bytes are unsupported, not estimated from retained heap. Enqueue-only allocations are also unsupported; completed-delivery workloads measure the full worker pipeline. +- Memory passes sample RSS at the specification's operation checkpoints and before/after work. Linux uses `/proc/self/status`; macOS uses `ps`. Linux lifetime peak RSS uses `VmHWM`; it is explicitly unsupported on macOS. `ObjectSpace.memsize_of_all` reports retained live-object bytes after settling, not total allocator arenas or cumulative allocation volume. Snapshot overhead is part of memory observation. GC runs before/after memory measurement, never inside the workload/checkpoints. +- Eight native threads perform the load scenario under Ruby's GVL; this does not promise eight parallel CPU cores. Load checkpoints join each 32-operation group. +- A coordinator-owned loopback receiver exercises the normal SDK HTTP, batching, byte-size splitting and background-worker paths. Receiver CPU, allocations and retained fixtures are outside the measured Ruby process. Refresh elapsed/allocations include the worker's loopback control request selecting the next generation. No production service or credentials are used. +- Public local-only flag getters and reload APIs are used, with flag-called events disabled. The single-flag getter's deprecation warning occurs during warmup. Capture uses public `batch_size: 100000`, `max_queue_size: 100001`, `flush_interval_seconds: 3600` and `compress_request: false`. Native byte-size batching remains active. Warmup deliveries are verified before counters reset; sequences, payloads, users and final delivery are validated without a rescuing flush for delivery workloads. + +Workers have a 120-second deadline. After two timeouts a coordinator finishes the current pair, saves partial diagnostics and fails. Each CI shard has a 12-minute job and a 10-minute measurement deadline, reserving time for upload. Collection has five minutes and rejects incomplete shard coverage. Run duration must be assessed from hosted CI, not concurrent local measurements. + +The shared reporter validates raw result schema v1 and produces readable tables plus `comparison.json`. Its comment shortlist requires at least a 5% median change and all six pairs agreeing in direction; it is an informational noise filter, not statistical significance. Missing and unsupported counters remain explicit. The read-only producer workflow is separate from the trusted `workflow_run` publisher; automatic PR comments become active only once that caller workflow is on the default branch. + ## Rails package The `posthog-rails` package has its own package-specific guide in [posthog-rails/CONTRIBUTING.md](posthog-rails/CONTRIBUTING.md). diff --git a/benchmarks/.gitignore b/benchmarks/.gitignore new file mode 100644 index 0000000..2b42731 --- /dev/null +++ b/benchmarks/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +results/ diff --git a/benchmarks/check_native.py b/benchmarks/check_native.py new file mode 100644 index 0000000..036f57e --- /dev/null +++ b/benchmarks/check_native.py @@ -0,0 +1,97 @@ +"""Run native controls plus regression checks through an installed SDK and loopback receiver.""" + +import argparse +import json +import subprocess +import tempfile +from pathlib import Path + +from fixtures import workloads +from run import HERE, TIMEOUT, WORKER, run_worker, worker_env +from server import BenchmarkServer + + +def main(installation): + with tempfile.TemporaryDirectory(prefix="ruby-controls-") as scratch: + env = worker_env(installation, scratch) + subprocess.run( + ["ruby", str(HERE / "controls.rb")], + env=env, + cwd=scratch, + check=True, + timeout=TIMEOUT, + ) + capture = next( + s for s in workloads(True) if s["id"] == "capture.drain.steady.32" + ) + code = """require ARGV[0] +PostHog::Logging.logger = Logger.new(File::NULL) +w = RubyBenchmark::Workload.new(JSON.parse(ARGV[1]), ARGV[2]) +begin + warmup = w.prepare(200) + raise 'Warmup size changed' unless warmup == 160 + raise 'Warmup was not reset' unless w.control('state')['delivered'].zero? + # Simulate a flush that returns without delivery at the benchmark seam, not inside the SDK. + def w.flush; end + w.run(200) + begin + w.complete + raise 'Completed-delivery verification rescued a missing flush' + rescue RuntimeError => error + raise unless error.message == 'Capture payload, identity or sequence mismatch' + end +ensure + w.close +end +puts 'Incomplete delivery rejected without a rescuing verification flush' +""" + with BenchmarkServer(capture) as server: + subprocess.run( + ["ruby", "-e", code, str(WORKER), json.dumps(capture), server.url], + env=env, + cwd=scratch, + check=True, + timeout=TIMEOUT, + ) + refresh = next(s for s in workloads(True) if s["mode"] == "refresh") + code = """require ARGV[0] +PostHog::Logging.logger = Logger.new(File::NULL) +w = RubyBenchmark::Workload.new(JSON.parse(ARGV[1]), ARGV[2]) +begin + w.prepare(4) + w.reload(1) + w.verify + w.reload(2) + w.verify +ensure + w.close +end +puts 'Changing refresh generations verified for all users' +""" + with BenchmarkServer(refresh) as server: + subprocess.run( + ["ruby", "-e", code, str(WORKER), json.dumps(refresh), server.url], + env=env, + cwd=scratch, + check=True, + timeout=TIMEOUT, + ) + # Real child timeout and capped diagnostic output; no SDK state manipulation. + sleeper = Path(scratch) / "sleeper.rb" + sleeper.write_text("STDERR.write('x' * 100000); STDERR.flush; sleep 60\n") + import run + + original = run.WORKER + try: + run.WORKER = sleeper + raw = run_worker(installation, refresh, "memory", timeout=0.5) + assert raw["status"] == "timeout" and 0 < len(raw["stderr"]) <= 65536 + finally: + run.WORKER = original + print("Real worker timeout classified with capped logs") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("installation", type=Path) + main(parser.parse_args().installation.resolve()) diff --git a/benchmarks/controls.rb b/benchmarks/controls.rb new file mode 100644 index 0000000..4f49c45 --- /dev/null +++ b/benchmarks/controls.rb @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +require_relative 'worker' + +# Native counter and fixture controls run against the installed public SDK entry point. +module RubyBenchmarkControls + def self.check(condition, message) + raise message unless condition + end + + def self.run + users = RubyBenchmark::USERS + check(users.length == 16, 'Expected 16 users') + check(users.map { |u| u[:id] }.uniq.length == 16, 'User identities must be unique') + users.each do |user| + check(RubyBenchmark.expected(0, user), 'Generation zero boolean') + check(!RubyBenchmark.expected(0, user, 1), 'Odd generation boolean') + check(!RubyBenchmark.expected(9, user), 'Inactive flag') + check(!RubyBenchmark.expected(8, user), 'No-match flag') + check(!RubyBenchmark.expected(5, user, 1), 'Odd generation cohort') + check(%w[control test].include?(RubyBenchmark.expected(2, user)), 'Variant') + check(RubyBenchmark.expected(6, user) == RubyBenchmark.expected(7, user), 'Early and late match equivalence') + end + before = GC.stat(:total_allocated_objects) + objects = Array.new(1000) { Object.new } + allocated = GC.stat(:total_allocated_objects) - before + check(allocated >= objects.length, 'Cumulative object counter must include live allocations') + nil + RubyBenchmark.settle + check(GC.stat(:total_allocated_objects) >= before + 1000, 'GC must not reset cumulative allocations') + snapshot = RubyBenchmark.snapshot + check(snapshot[:rss].positive? && snapshot[:heap_used].positive?, 'Native memory counters') + start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) + sleep(0.002) + check(Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) > start, 'Monotonic clock') + check(RubyBenchmark.describe[:entry] == Gem.loaded_specs.fetch('posthog-ruby').full_gem_path, + 'Installed gem identity') + puts 'Native fixture, allocation, memory, clock and installed-entry controls passed' + end +end + +RubyBenchmarkControls.run if $PROGRAM_NAME == __FILE__ diff --git a/benchmarks/fixtures.py b/benchmarks/fixtures.py new file mode 100644 index 0000000..5c81087 --- /dev/null +++ b/benchmarks/fixtures.py @@ -0,0 +1,216 @@ +"""Deterministic fixtures for sdk-benchmarks/v1.""" + +import hashlib +import json +from typing import Any + +CASES = [ + "boolean", + "rollout", + "variant", + "property", + "regex", + "cohort", + "early", + "late", + "no-match", + "inactive", +] +USERS = [ + { + "distinct_id": f"benchmark-user-{i}", + "person_properties": { + "plan": "free" if i % 2 else "pro", + "email": f"user-{i}@example.test", + "score": i, + }, + } + for i in range(16) +] + + +def prop(key, value, operator="exact"): + return {"key": key, "value": value, "operator": operator, "type": "person"} + + +def group(properties=None, rollout=100): + return {"properties": properties or [], "rollout_percentage": rollout} + + +def definitions(count, generation=0): + flags = [] + for i in range(count): + kind = CASES[i % len(CASES)] + match = group([prop("plan", ["pro"])]) + miss = group([prop("plan", ["never"])]) + groups = [group()] + if kind == "rollout": + groups = [group(rollout=50)] + elif kind == "property": + groups = [match] + elif kind == "regex": + groups = [ + group( + [ + prop( + "email", + f"^(user|refresh-{generation})-[0-9]+@example[.]test$", + "regex", + ) + ] + ) + ] + elif kind == "cohort": + groups = [group([{"key": "id", "value": 1, "type": "cohort"}])] + elif kind == "early": + groups = [match] + [miss] * 7 + elif kind == "late": + groups = [miss] * 7 + [match] + elif kind == "no-match": + groups = [miss] * 8 + filters: dict[str, Any] = {"groups": groups} + if kind == "variant": + filters["multivariate"] = { + "variants": [ + {"key": "control", "rollout_percentage": 50}, + {"key": "test", "rollout_percentage": 50}, + ] + } + flags.append( + { + "id": i + 1, + "key": f"{kind}-{i}", + "name": f"{kind}-{i}", + "active": kind != "inactive" + and (kind != "boolean" or generation % 2 == 0), + "filters": filters, + } + ) + return { + "flags": flags, + "group_type_mapping": {}, + "cohorts": { + "1": { + "type": "AND", + "values": [ + prop("plan", ["pro"]), + prop("score", 100 if generation % 2 else -1, "gt"), + ], + }, + }, + } + + +def bucket(key, distinct_id, salt=""): + return int( + hashlib.sha1(f"{key}.{distinct_id}{salt}".encode()).hexdigest()[:15], 16 + ) / (2**60 - 1) + + +def expected_value(index, user, generation=0): + kind = CASES[index % len(CASES)] + key = f"{kind}-{index}" + if kind == "rollout": + return bucket(key, user["distinct_id"]) < 0.5 + if kind == "variant": + return ( + "control" if bucket(key, user["distinct_id"], "variant") < 0.5 else "test" + ) + if kind == "boolean": + return generation % 2 == 0 + if kind == "cohort" and generation % 2: + return False + if kind in ("property", "cohort", "early", "late"): + return user["person_properties"]["plan"] == "pro" + return kind not in ("inactive", "no-match") + + +def expected_flags(count, user, generation=0): + return { + f"{CASES[i % len(CASES)]}-{i}": expected_value(i, user, generation) + for i in range(count) + } + + +def digest(value): + data = ( + value + if isinstance(value, bytes) + else json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + ) + return hashlib.sha256(data).hexdigest() + + +def workloads(smoke=False): + return [ + *[ + { + "id": f"flags.single.{kind}", + "mode": "single", + "flag_count": 10, + "flag_index": i, + "operations": 160 if smoke else 16000, + "allocation_operations": 160 if smoke else 2000, + "concurrency": 1, + "passes": ["timing", "allocation"], + } + for i, kind in enumerate(CASES[:9]) + ], + *[ + { + "id": f"flags.all.{count}", + "mode": "all", + "flag_count": count, + "operations": 16 if smoke else max(160, 16000 // count), + "allocation_operations": 16 if smoke else max(16, 1600 // count), + "concurrency": 1, + "passes": ["timing", "allocation"], + } + for count in (10, 100, 1000) + ], + { + "id": "flags.load.1000", + "mode": "all", + "flag_count": 1000, + "operations": 32 if smoke else 1600, + "concurrency": 8, + "passes": ["memory"], + }, + { + "id": "flags.refresh.1000", + "mode": "refresh", + "flag_count": 1000, + "operations": 4 if smoke else 40, + "concurrency": 1, + "passes": ["memory"], + }, + *[ + { + "id": f"capture.enqueue.{size}", + "mode": "capture", + "payload_bytes": size, + "operations": 200 if smoke else 1000, + "completion_boundary": "api_return", + "concurrency": 1, + "passes": ["timing"], + } + for size in (32, 4096) + ], + *[ + { + "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/prepare.py b/benchmarks/prepare.py new file mode 100644 index 0000000..e42ba86 --- /dev/null +++ b/benchmarks/prepare.py @@ -0,0 +1,85 @@ +"""Build and install each SDK gem separately; workers never load a checkout.""" + +import argparse +import json +import os +import subprocess +from pathlib import Path + + +def prepare(directory, output): + directory, output = directory.resolve(), output.resolve() + # An old installed gem with the same version must not mask a new build. + output.mkdir(parents=True, exist_ok=False) + env = { + **os.environ, + "BUNDLE_GEMFILE": str(directory / "Gemfile"), + "BUNDLE_FROZEN": "true", + } + subprocess.run( + ["bundle", "install"], + cwd=directory, + env=env, + check=True, + timeout=120, + ) + dependency = json.loads( + subprocess.check_output( + [ + "bundle", + "exec", + "ruby", + "-rjson", + "-rconcurrent", + "-e", + 'puts JSON.generate({version: Concurrent::VERSION, gem_path: Gem.loaded_specs.fetch("concurrent-ruby").base_dir})', + ], + cwd=directory, + env=env, + text=True, + timeout=120, + ) + ) + artifact = output / "sdk.gem" + subprocess.run( + ["gem", "build", "posthog-ruby.gemspec", "--output", str(artifact)], + cwd=directory, + check=True, + timeout=120, + ) + gems = output / "gems" + subprocess.run( + [ + "gem", + "install", + str(artifact), + "--local", + "--ignore-dependencies", + "--no-document", + "--install-dir", + str(gems), + ], + cwd=output, + check=True, + timeout=120, + ) + specs = list((gems / "specifications").glob("posthog-ruby-*.gemspec")) + if len(specs) != 1: + raise ValueError("Expected one installed SDK gem") + version = specs[0].stem.removeprefix("posthog-ruby-") + info = { + "gem_home": str(gems), + "gem_path": dependency["gem_path"], + "sdk_version": version, + "concurrent_version": dependency["version"], + "artifact": str(artifact), + } + (output / "installation.json").write_text(json.dumps(info, indent=2) + "\n") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("directory", type=Path) + parser.add_argument("output", type=Path) + options = parser.parse_args() + prepare(options.directory, options.output) diff --git a/benchmarks/run.py b/benchmarks/run.py new file mode 100644 index 0000000..c0192d1 --- /dev/null +++ b/benchmarks/run.py @@ -0,0 +1,422 @@ +"""Paired Ruby SDK measurements using head-side fixtures and isolated installed gems.""" + +import argparse +import json +import math +import os +import re +import signal +import subprocess +import sys +import tempfile +import threading +import uuid +from pathlib import Path + +from fixtures import USERS, definitions, digest, workloads +from server import BenchmarkServer + +HERE = Path(__file__).resolve().parent +WORKER = HERE / "worker.rb" +TIMEOUT = 120 +MAX_TIMEOUTS = 2 + + +def git(directory, *args): + return subprocess.check_output( + ["git", "-C", str(directory), *args], text=True + ).strip() + + +def source_digest(): + return digest( + { + p.name: digest(p.read_bytes()) + for p in sorted(HERE.iterdir()) + if p.suffix in (".rb", ".py", ".json") + } + ) + + +def load_shards(): + manifest = json.loads((HERE / "shards.json").read_text()) + if manifest.get("manifest_version") != 1 or manifest.get("sdk") != "posthog-ruby": + raise ValueError("Invalid shard manifest") + shards = {} + seen = set() + for shard in manifest["shards"]: + identity, members = shard["id"], shard["workloads"] + if ( + not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", identity) + or identity in shards + or not members + ): + raise ValueError("Invalid shard identity") + if len(set(members)) != len(members) or seen.intersection(members): + raise ValueError("Duplicate shard workloads") + seen.update(members) + shards[identity] = members + if not 0 < len(shards) <= 16 or any( + seen != {w["id"] for w in workloads(smoke)} for smoke in (True, False) + ): + raise ValueError("Shard manifest must partition full and smoke inventories") + return shards + + +def worker_env(installation, directory): + info = json.loads((installation / "installation.json").read_text()) + env = { + key: value + for key, value in os.environ.items() + if not key.startswith(("BUNDLE_", "RUBY", "GEM_")) + } + for key in ( + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", + "http_proxy", + "https_proxy", + "all_proxy", + "POSTHOG_RELEASE_ID", + ): + env.pop(key, None) + env.update( + GEM_HOME=info["gem_home"], + GEM_PATH=os.pathsep.join([info["gem_home"], info["gem_path"]]), + RUBYOPT="", + BENCHMARK_SDK_VERSION=info["sdk_version"], + BENCHMARK_CONCURRENT_VERSION=info["concurrent_version"], + TMPDIR=str(directory), + HOME=str(directory), + ) + return env + + +def run_worker(installation, spec, measurement_pass, timeout=TIMEOUT): + with tempfile.TemporaryDirectory( + prefix="ruby-benchmark-" + ) as scratch, BenchmarkServer(spec) as server: + scratch = Path(scratch) + input_path, output = scratch / "spec.json", scratch / "result.json" + input_path.write_text(json.dumps({**spec, "pass": measurement_pass})) + process = subprocess.Popen( + ["ruby", str(WORKER), str(input_path), server.url, str(output)], + cwd=scratch, + env=worker_env(installation, scratch), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + ) + buffers = {"stdout": bytearray(), "stderr": bytearray()} + + def read_log(stream, target): + with stream: + while chunk := stream.read(4096): + target.extend(chunk) + del target[:-65536] + + readers = [ + threading.Thread( + target=read_log, args=(getattr(process, name), buffer), daemon=True + ) + for name, buffer in buffers.items() + ] + for reader in readers: + reader.start() + timed_out = False + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + os.killpg(process.pid, signal.SIGKILL) + process.wait() + for reader in readers: + reader.join(timeout=2) + logs = { + name: data.decode("utf-8", errors="replace") + for name, data in buffers.items() + } + if process.returncode == 0 and output.exists() and not timed_out: + if output.stat().st_size > 2 * 1024 * 1024: + raise ValueError("Worker result exceeds size limit") + result = json.loads(output.read_text()) + operations = ( + spec.get("allocation_operations", spec["operations"]) + if measurement_pass == "allocation" + else spec["operations"] + ) + if ( + result.get("pass") != measurement_pass + or result.get("operations") != operations + or result.get("status") != "ok" + ): + raise ValueError("Worker result identity differs") + return {**result, **logs} + return { + "status": "timeout" if timed_out else "error", + "pass": measurement_pass, + "code": process.returncode, + "error": ( + "Worker exceeded deadline" + if timed_out + else "Worker did not complete; exit alone does not establish OOM" + ), + **logs, + } + + +def document_for(directory, installation, role, pair_id, suite, shard): + with tempfile.TemporaryDirectory(prefix="ruby-benchmark-describe-") as scratch: + execution = json.loads( + subprocess.check_output( + ["ruby", str(WORKER), "--describe"], + cwd=scratch, + env=worker_env(installation, scratch), + text=True, + timeout=TIMEOUT, + ) + ) + # Fingerprint the installed build, independently of package version. + installed = Path(execution["entry"]) + execution.update( + pair_id=pair_id, + runtime_flags=[], + build_digest=digest( + { + str(p.relative_to(installed)): digest(p.read_bytes()) + for p in sorted((installed / "lib").rglob("*.rb")) + } + ), + dependency_lock_digest=digest((directory / "Gemfile.lock").read_bytes()), + runner_class=( + "github-hosted-or-self-hosted" if os.getenv("RUNNER_NAME") else "local" + ), + runner_image=os.getenv("ImageVersion"), + resource_limits={ + "worker_timeout_ms": TIMEOUT * 1000, + "max_worker_timeouts": MAX_TIMEOUTS, + "process_rss_limit_bytes": None, + }, + ) + if shard: + execution["shard_id"] = shard + return { + "schema_version": 1, + "sdk": "posthog-ruby", + "revision": { + "role": role, + "commit": git(directory, "rev-parse", "HEAD"), + "dirty": bool(git(directory, "status", "--porcelain")), + }, + "suite": suite, + "execution": execution, + "benchmarks": [], + } + + +def append_sample(record, raw, repetition): + record["runs"].append({"repetition": repetition, **raw}) + if raw["status"] != "ok": + record["status"] = raw["status"] + return + + def add(name, unit, method, value): + if not math.isfinite(value) or value < 0: + raise ValueError(f"Invalid metric: {name}") + metric = next((m for m in record["metrics"] if m["id"] == name), None) + if metric is None: + metric = { + "id": name, + "unit": unit, + "method": method, + "quality": "measured", + "direction": "lower", + "status": "ok", + "samples": [], + } + record["metrics"].append(metric) + metric["samples"].append( + {"repetition": repetition, "value": value, "operations": raw["operations"]} + ) + + def unsupported(name, unit, reason): + if not any(m["id"] == name for m in record["metrics"]): + record["metrics"].append( + {"id": name, "unit": unit, "status": "unsupported", "reason": reason} + ) + + if raw["pass"] == "timing": + add( + "elapsed_time_per_op", + "ns/op", + "Process.clock_gettime.CLOCK_MONOTONIC", + raw["elapsed_time_per_op"], + ) + enqueue = record["parameters"]["mode"] == "capture" + add( + "caller_thread_cpu_time_per_op" if enqueue else "process_cpu_time_per_op", + "ns/op", + ( + "Process.clock_gettime.CLOCK_THREAD_CPUTIME_ID" + if enqueue + else "Process.clock_gettime.CLOCK_PROCESS_CPUTIME_ID" + ), + raw["cpu_time_per_op"], + ) + if enqueue: + for name, unit in ( + ("allocated_bytes_per_op", "bytes/op"), + ("allocations_per_op", "count/op"), + ): + unsupported( + name, + unit, + "Background serialization has no public enqueue completion boundary; use completed-delivery workloads for process-wide allocation counts.", + ) + elif raw["pass"] == "allocation": + add( + "allocations_per_op", + "count/op", + "GC.stat.total_allocated_objects; all worker threads; Ruby objects excluding native malloc", + raw["allocated_objects"] / raw["operations"], + ) + unsupported( + "allocated_bytes_per_op", + "bytes/op", + "Ruby has no cumulative allocated-byte counter here; retained ObjectSpace bytes are not allocation volume.", + ) + elif raw["pass"] == "memory": + for name, value in ( + ("idle_process_rss", raw["before"]["rss"]), + ("observed_workload_peak_rss", raw["observed_peak_rss"]), + ("post_settling_process_rss", raw["after_settling"]["rss"]), + ): + add( + name, + "bytes", + "proc.status.VmRSS or ps.rss; operation checkpoints", + value, + ) + add( + "post_settling_heap_used", + "bytes", + "ObjectSpace.memsize_of_all; retained live objects, not allocator arena size", + raw["after_settling"]["heap_used"], + ) + if raw["process_lifetime_peak_rss"] is not None: + add( + "process_lifetime_peak_rss", + "bytes", + "proc.status.VmHWM; includes setup and warmup", + raw["process_lifetime_peak_rss"], + ) + else: + unsupported( + "process_lifetime_peak_rss", + "bytes", + "No process-lifetime RSS high-water counter on this platform; not inferred from checkpoints.", + ) + + +def main(args=None): + parser = argparse.ArgumentParser(description=__doc__) + for name in ("base", "head", "build_directory", "output"): + parser.add_argument("--" + name.replace("_", "-"), type=Path, required=True) + parser.add_argument("--smoke", action="store_true") + parser.add_argument("--repetitions", type=int, default=6, choices=range(1, 31)) + selection = parser.add_mutually_exclusive_group() + selection.add_argument("--shard") + selection.add_argument("--filter", default="") + options = parser.parse_args(args) + shards = load_shards() + if options.shard and options.shard not in shards: + parser.error("Unknown shard") + specs = ( + [s for s in workloads(options.smoke) if s["id"] in shards[options.shard]] + if options.shard + else [s for s in workloads(options.smoke) if s["id"].startswith(options.filter)] + ) + if not specs: + parser.error("No matching workloads") + suite = { + "id": "ruby-sdk-performance", + "version": 2, + "specification": "sdk-benchmarks/v1", + "source_commit": git(HERE, "rev-parse", "HEAD"), + "source_digest": source_digest(), + "profile": "smoke" if options.smoke else "full", + "repetitions": options.repetitions, + "transport": "coordinator-owned loopback HTTP receiver; uncompressed; native batch byte limits retained", + "capture": { + "batch_size": 100000, + "max_queue_size": 100001, + "flush_interval_seconds": 3600, + }, + "warmup": "two refreshes; otherwise min(operations, 32 for 1000 flags or 160); verified reset and generation zero restored", + "settling": "two GC.start calls separated by Thread.pass; no forced GC in measured work", + "memory_sampling": "after every 32 load operations, each refresh, each 100 captures, and final work; ObjectSpace and RSS snapshots", + "allocation_scope": "GC.stat Ruby object allocations across worker threads; excludes coordinator receiver and native malloc", + } + pair_id = str(uuid.uuid4()) + directories = {"base": options.base.resolve(), "head": options.head.resolve()} + builds = {role: options.build_directory.resolve() / role for role in directories} + documents = { + role: document_for(directory, builds[role], role, pair_id, suite, options.shard) + for role, directory in directories.items() + } + for key in ("runtime", "runtime_version", "measurement_tools", "os", "arch"): + if documents["base"]["execution"][key] != documents["head"]["execution"][key]: + raise ValueError(f"Base/head {key} differ") + options.output.mkdir(parents=True, exist_ok=True) + + def save(): + for role, document in documents.items(): + temporary = options.output / f"{role}.json.tmp" + temporary.write_text(json.dumps(document, allow_nan=False, indent=2) + "\n") + temporary.replace(options.output / f"{role}.json") + + failures = timeouts = 0 + for spec in specs: + records = {} + for role in directories: + record = { + "id": spec["id"], + "parameters": spec, + "fixture_digest": digest( + { + "spec": spec, + "users": USERS, + "definitions": ( + definitions(spec["flag_count"]) + if "flag_count" in spec + else None + ), + } + ), + "status": "ok", + "metrics": [], + "runs": [], + } + records[role] = record + documents[role]["benchmarks"].append(record) + for repetition in range(options.repetitions): + for measurement_pass in spec["passes"]: + for role in ("head", "base") if repetition % 2 else ("base", "head"): + raw = run_worker(builds[role], spec, measurement_pass) + append_sample(records[role], raw, repetition) + failures += raw["status"] != "ok" + timeouts += raw["status"] == "timeout" + print( + f'{spec["id"]} {measurement_pass} {repetition + 1}/{options.repetitions} {role}: {raw["status"]}', + flush=True, + ) + save() + if timeouts >= MAX_TIMEOUTS: + return 1 + if any(record["status"] != "ok" for record in records.values()): + break + return int(failures > 0) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/server.py b/benchmarks/server.py new file mode 100644 index 0000000..b119c2f --- /dev/null +++ b/benchmarks/server.py @@ -0,0 +1,159 @@ +"""Bounded loopback receiver owned by the coordinator, never the measured worker. + +The SDK has no public transport injection option. HTTP preserves its normal +loading, serialization, background-consumer and flush paths. +""" + +import json +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +from urllib.parse import urlsplit + +from fixtures import USERS, definitions + + +class BenchmarkServer: + def __init__(self, spec, batch_delay=0): + self.spec = spec + 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": [], + } + owner = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def reply(self, status, body): + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + with owner.lock: + path = urlsplit(self.path).path.rstrip("/") + if path == "/flags/definitions": + owner.state["definitions"] += 1 + self.reply(200, owner.body) + elif path == "/__benchmark/state": + self.reply(200, json.dumps(owner.state).encode()) + else: + self.unexpected(path) + + def unexpected(self, path): + if len(owner.state["unexpected"]) < 20: + owner.state["unexpected"].append(path) + self.reply(404, b'{"error":"unexpected benchmark request"}') + + def do_POST(self): + with owner.lock: + path = urlsplit(self.path).path.rstrip("/") + try: + size = int(self.headers.get("Content-Length", "0")) + if not 0 < size <= 20 * 1024 * 1024: + raise ValueError("Invalid body length") + body = json.loads(self.rfile.read(size)) + 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) + or not 0 <= generation <= 1000 + ): + raise ValueError("Invalid generation") + owner.body = json.dumps( + definitions(spec["flag_count"], generation) + ).encode() + self.reply(200, b"{}") + elif path == "/batch": + if batch_delay: + time.sleep(batch_delay) + batch = body["batch"] + if not isinstance(batch, list): + raise ValueError("Invalid batch") + owner.state["requests"] += 1 + owner.state["delivered"] += len(batch) + 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) + except (ValueError, KeyError, TypeError, AttributeError): + owner.state["invalid_events"] += 1 + self.reply(400, b'{"error":"invalid benchmark body"}') + + self.http = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.thread = threading.Thread( + target=self.http.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True + ) + self.url = f"http://127.0.0.1:{self.http.server_port}" + + def __enter__(self): + self.thread.start() + return self + + def __exit__(self, *_): + self.http.shutdown() + self.http.server_close() + self.thread.join() diff --git a/benchmarks/shards.json b/benchmarks/shards.json new file mode 100644 index 0000000..8a17e28 --- /dev/null +++ b/benchmarks/shards.json @@ -0,0 +1,39 @@ +{ + "manifest_version": 1, + "sdk": "posthog-ruby", + "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 new file mode 100644 index 0000000..0ab29fa --- /dev/null +++ b/benchmarks/test_benchmarks.py @@ -0,0 +1,343 @@ +"""Coordinator, fixture and real public-entry controls. No external service required.""" + +import copy +import json +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch +from urllib.request import Request, urlopen + +import fixtures +import run +from server import BenchmarkServer + +HERE = Path(__file__).resolve().parent + + +class Controls(unittest.TestCase): + def test_inventory_and_partition(self): + shards = run.load_shards() + self.assertEqual(len(shards), 4) + for smoke in (True, False): + specs = fixtures.workloads(smoke) + self.assertEqual(len(specs), 20) + self.assertEqual(len({s["id"] for s in specs}), 20) + self.assertEqual( + set().union(*(set(w) for w in shards.values())), + {s["id"] for s in specs}, + ) + self.assertEqual(len([s for s in specs if s["mode"] == "single"]), 9) + for spec in specs: + ( + self.assertEqual(spec["operations"], 200 if smoke else 1000) + if "payload_bytes" in spec + else None + ) + if spec["mode"] == "drain": + self.assertEqual( + spec["flush_every"], + ( + 100 + if spec["drain_strategy"] == "steady" + else spec["operations"] + ), + ) + self.assertEqual(spec["passes"], ["timing", "allocation", "memory"]) + + def test_flags_fixtures(self): + doc = fixtures.definitions(1000) + self.assertEqual(len(doc["flags"]), 1000) + self.assertEqual(len(fixtures.USERS), 16) + self.assertEqual(doc["flags"][9]["active"], False) + self.assertEqual(len(doc["flags"][6]["filters"]["groups"]), 8) + self.assertEqual(len(doc["flags"][7]["filters"]["groups"]), 8) + self.assertNotEqual( + fixtures.definitions(1000, 0), fixtures.definitions(1000, 1) + ) + self.assertEqual( + fixtures.definitions(1000, 0), + fixtures.definitions(1000, 2) + | {"flags": fixtures.definitions(1000, 0)["flags"]}, + ) + + def test_source_identity_includes_manifest(self): + before = run.source_digest() + original = (HERE / "shards.json").read_bytes() + real = Path.read_bytes + with patch.object( + Path, + "read_bytes", + lambda p: original + b" " if p.name == "shards.json" else real(p), + ): + self.assertNotEqual(run.source_digest(), before) + + def test_bad_manifests_rejected(self): + manifest = json.loads((HERE / "shards.json").read_text()) + for mutate in ( + lambda m: m["shards"][0]["workloads"].pop(), + lambda m: m["shards"][1]["workloads"].append( + m["shards"][0]["workloads"][0] + ), + lambda m: m["shards"][0].update(id="../unsafe"), + ): + changed = copy.deepcopy(manifest) + mutate(changed) + with patch.object( + Path, "read_text", return_value=json.dumps(changed) + ), self.assertRaises(ValueError): + run.load_shards() + + def test_metrics_keep_failures_and_unsupported_distinct(self): + record = { + "status": "ok", + "parameters": {"mode": "all"}, + "runs": [], + "metrics": [], + } + run.append_sample( + record, + { + "status": "ok", + "pass": "allocation", + "operations": 10, + "allocated_objects": 100, + }, + 0, + ) + values = {m["id"]: m for m in record["metrics"]} + self.assertEqual(values["allocations_per_op"]["samples"][0]["value"], 10) + self.assertEqual(values["allocated_bytes_per_op"]["status"], "unsupported") + run.append_sample(record, {"status": "timeout", "pass": "allocation"}, 1) + self.assertEqual(record["status"], "timeout") + self.assertEqual(len(values["allocations_per_op"]["samples"]), 1) + + def test_invalid_counter_rejected(self): + record = { + "status": "ok", + "parameters": {"mode": "all"}, + "runs": [], + "metrics": [], + } + with self.assertRaises(ValueError): + run.append_sample( + record, + { + "status": "ok", + "pass": "allocation", + "operations": 1, + "allocated_objects": -1, + }, + 0, + ) + + def test_enqueue_uses_caller_thread_cpu(self): + record = { + "status": "ok", + "parameters": {"mode": "capture"}, + "runs": [], + "metrics": [], + } + run.append_sample( + record, + { + "status": "ok", + "pass": "timing", + "operations": 1, + "elapsed_time_per_op": 2, + "cpu_time_per_op": 1, + }, + 0, + ) + self.assertEqual( + {m["id"] for m in record["metrics"]}, + { + "elapsed_time_per_op", + "caller_thread_cpu_time_per_op", + "allocations_per_op", + "allocated_bytes_per_op", + }, + ) + + def test_timeout_finishes_pair_and_stops(self): + doc = { + "schema_version": 1, + "execution": { + k: "same" + for k in ( + "runtime", + "runtime_version", + "measurement_tools", + "os", + "arch", + ) + }, + "benchmarks": [], + } + calls = [] + + def fake_worker(installation, spec, measurement_pass): + calls.append(installation.name) + return {"status": "timeout", "pass": measurement_pass} + + with tempfile.TemporaryDirectory() as t, patch.object( + run, "document_for", side_effect=lambda *a: copy.deepcopy(doc) + ), patch.object(run, "run_worker", side_effect=fake_worker): + code = run.main( + [ + "--base", + t, + "--head", + t, + "--build-directory", + t, + "--output", + t, + "--shard", + "flags-single", + ] + ) + self.assertEqual(code, 1) + self.assertEqual(calls, ["base", "head"]) + saved = json.loads((Path(t) / "head.json").read_text()) + self.assertEqual(saved["benchmarks"][0]["status"], "timeout") + + def test_alternating_pairs(self): + doc = { + "execution": { + k: "same" + for k in ( + "runtime", + "runtime_version", + "measurement_tools", + "os", + "arch", + ) + }, + "benchmarks": [], + } + calls = [] + + def worker(installation, spec, measurement_pass): + calls.append(installation.name) + return { + "status": "ok", + "pass": measurement_pass, + "operations": 1, + "elapsed_time_per_op": 1, + "cpu_time_per_op": 1, + "allocated_objects": 1, + } + + with tempfile.TemporaryDirectory() as t, patch.object( + run, "document_for", side_effect=lambda *a: copy.deepcopy(doc) + ), patch.object(run, "run_worker", side_effect=worker): + self.assertEqual( + run.main( + [ + "--base", + t, + "--head", + t, + "--build-directory", + t, + "--output", + t, + "--filter", + "flags.single.boolean", + "--repetitions", + "2", + ] + ), + 0, + ) + self.assertEqual( + calls, ["base", "head", "base", "head", "head", "base", "head", "base"] + ) + + def test_describe_is_bounded(self): + with tempfile.TemporaryDirectory() as t, patch.object( + run, "worker_env", return_value={} + ), patch.object( + run.subprocess, + "check_output", + side_effect=subprocess.TimeoutExpired("ruby", run.TIMEOUT), + ) as command: + with self.assertRaises(subprocess.TimeoutExpired): + run.document_for(Path(t), Path(t), "head", "pair", {}, None) + self.assertEqual(command.call_args.kwargs["timeout"], run.TIMEOUT) + + def test_unknown_shard_and_filter_conflict(self): + with self.assertRaises(SystemExit): + run.main( + [ + "--base", + ".", + "--head", + ".", + "--build-directory", + ".", + "--output", + ".", + "--shard", + "unknown", + ] + ) + with self.assertRaises(SystemExit): + run.main( + [ + "--base", + ".", + "--head", + ".", + "--build-directory", + ".", + "--output", + ".", + "--shard", + "capture", + "--filter", + "flags", + ] + ) + + def test_receiver_sequence_validation(self): + spec = next( + s for s in fixtures.workloads(True) if s["id"] == "capture.drain.burst.32" + ) + + def request(server, path, body=None): + req = Request( + server.url + path, + data=json.dumps(body).encode() if body is not None else None, + ) + with urlopen(req, timeout=5) as response: + return json.load(response) + + def event(i): + return { + "event": "benchmark", + "distinct_id": fixtures.USERS[i % 16]["distinct_id"], + "properties": {"sequence": i, "plan": "pro", "payload": "x" * 32}, + } + + with BenchmarkServer(spec) as server: + request(server, "/__benchmark/reset", {"operations": 32}) + request( + server, "/batch", {"batch": [event(i) for i in reversed(range(32))]} + ) + state = request(server, "/__benchmark/state") + self.assertEqual(state["missing_sequences"], 0) + self.assertEqual(state["invalid_events"], 0) + request(server, "/__benchmark/reset", {"operations": 32}) + # Balanced duplicate/loss preserves event/user totals but must fail. + request(server, "/batch", {"batch": [event(i) for i in range(16)] * 2}) + state = request(server, "/__benchmark/state") + self.assertEqual(state["delivered"], 32) + self.assertGreater(state["missing_sequences"], 0) + self.assertGreater(state["invalid_events"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/benchmarks/worker.rb b/benchmarks/worker.rb new file mode 100644 index 0000000..ec2c2be --- /dev/null +++ b/benchmarks/worker.rb @@ -0,0 +1,291 @@ +# frozen_string_literal: true + +require 'json' +require 'net/http' +require 'digest' +require 'objspace' +require 'logger' +require 'rbconfig' + +gem 'concurrent-ruby', ENV.fetch('BENCHMARK_CONCURRENT_VERSION') +gem 'posthog-ruby', ENV.fetch('BENCHMARK_SDK_VERSION') +require 'posthog' + +module RubyBenchmark + CASES = %w[boolean rollout variant property regex cohort early late no-match inactive].freeze + USERS = Array.new(16) do |i| + { id: "benchmark-user-#{i}", properties: { plan: i.even? ? 'pro' : 'free', + email: "user-#{i}@example.test", score: i } } + end.freeze + + def self.require_condition(condition, message) + raise message unless condition + end + + def self.expected(index, user, generation = 0) + kind = CASES[index % 10] + return generation.even? if kind == 'boolean' + return false if %w[inactive no-match].include?(kind) || (kind == 'cohort' && generation.odd?) + return user[:properties][:plan] == 'pro' if %w[property cohort early late].include?(kind) + + if %w[rollout variant].include?(kind) + salt = kind == 'variant' ? 'variant' : '' + bucket = Digest::SHA1.hexdigest("#{kind}-#{index}.#{user[:id]}#{salt}")[0, 15].to_i(16).to_f / ((2**60) - 1) + return bucket < 0.5 if kind == 'rollout' + + return bucket < 0.5 ? 'control' : 'test' + end + true + end + + def self.settle + GC.start + Thread.pass + GC.start + end + + def self.snapshot + # ObjectSpace reports retained live-object bytes, not cumulative allocation volume. + heap = ObjectSpace.memsize_of_all + rss = if File.file?('/proc/self/status') + File.read('/proc/self/status')[/^VmRSS:\s+(\d+)/, 1].to_i * 1024 + else + IO.popen(['ps', '-o', 'rss=', '-p', Process.pid.to_s], &:read).to_i * 1024 + end + require_condition(rss.positive?, 'RSS counter unavailable') + { rss: rss, heap_used: heap } + end + + def self.describe + { + runtime: RUBY_ENGINE, runtime_version: RUBY_VERSION, os: RbConfig::CONFIG['host_os'], + arch: RbConfig::CONFIG['host_cpu'], + cpu: if File.file?('/proc/cpuinfo') + File.read('/proc/cpuinfo')[/^model name\s*:\s*(.+)/, 1] || 'unknown' + else + IO.popen(['sysctl', '-n', 'machdep.cpu.brand_string'], &:read).strip + end, + kernel: IO.popen(['uname', '-r'], &:read).strip, + sdk_version: PostHog::VERSION, + entry: Gem.loaded_specs.fetch('posthog-ruby').full_gem_path, + packages: { 'concurrent-ruby' => Concurrent::VERSION }, + measurement_tools: { ruby: RUBY_DESCRIPTION, objectspace: 'stdlib' } + } + end + + class Workload + def initialize(spec, url) + @spec = spec + @url = url + @generation = 0 + @captured = 0 + @errors = [] + @client = PostHog::Client.new( + api_key: 'benchmark-project', secret_key: spec.key?('flag_count') ? 'benchmark-secret' : nil, + host: url, batch_size: 100_000, max_queue_size: 100_001, flush_interval_seconds: 3600, + feature_flags_polling_interval: 3600, compress_request: false, max_retries: 0, + feature_flag_request_max_retries: 0, disable_singleton_warning: true, + on_error: ->(status, error) { @errors << [status, error.to_s[0, 300]] if @errors.length < 10 } + ) + end + + def control(path, value = nil) + uri = URI("#{@url}/__benchmark/#{path}") + http = Net::HTTP.new(uri.host, uri.port, nil) + http.open_timeout = http.read_timeout = 5 + request = value ? Net::HTTP::Post.new(uri) : Net::HTTP::Get.new(uri) + request.body = JSON.generate(value) if value + response = http.request(request) + RubyBenchmark.require_condition(response.code == '200', 'Benchmark control failed') + JSON.parse(response.body) + end + + def reload(generation) + control('definitions', { generation: generation }) + @client.reload_feature_flags + @generation = generation + end + + def all_flags(user) + @client.get_all_flags(user[:id], person_properties: user[:properties], only_evaluate_locally: true) + end + + def single_flag(user) + index = @spec.fetch('flag_index') + @client.get_feature_flag( + "#{CASES[index]}-#{index}", user[:id], person_properties: user[:properties], + only_evaluate_locally: true, send_feature_flag_events: false + ) + end + + def prepare_events(operations) + @events = Array.new(operations) do |i| + { event: 'benchmark', distinct_id: USERS[i % 16][:id], + properties: { payload: 'x' * @spec.fetch('payload_bytes'), plan: 'pro', sequence: i } } + end + end + + def run(operations) + mode = @spec.fetch('mode') + if mode == 'refresh' + operations.times do |i| + reload(i + 1) + yield(i + 1) if block_given? + end + elsif %w[capture drain].include?(mode) + operations.times do |i| + RubyBenchmark.require_condition(@client.capture(@events[i]), 'Capture rejected') + @captured += 1 + flush if mode == 'drain' && @spec['drain_strategy'] == 'steady' && + (((i + 1) % @spec['flush_every']).zero? || i + 1 == operations) + yield(i + 1) if block_given? && ((i + 1) % 100).zero? + end + flush if mode == 'drain' && @spec['drain_strategy'] == 'burst' + elsif @spec.fetch('concurrency') > 1 + # Join in deterministic 32-operation groups, preserving native GVL-constrained threads. + operations.times.each_slice(32) do |positions| + threads = positions.each_slice(4).map do |slice| + Thread.new { slice.each { |i| all_flags(USERS[i % 16]) } } + end + threads.each(&:value) + yield(positions.last + 1) if block_given? + end + else + operations.times do |i| + mode == 'single' ? single_flag(USERS[i % 16]) : all_flags(USERS[i % 16]) + end + end + end + + def flush + RubyBenchmark.require_condition(@client.flush(timeout: 10), 'Public flush timed out') + end + + def verify + RubyBenchmark.require_condition(@errors.empty?, "SDK callback errors: #{@errors.inspect}") + state = control('state') + RubyBenchmark.require_condition(state.fetch('unexpected').empty?, 'Unexpected network request') + if @spec.key?('flag_count') + RubyBenchmark.require_condition(@client.feature_flags_loaded?, 'Flag definitions not loaded') + USERS.each do |user| + expected = @spec['flag_count'].times.to_h do |i| + ["#{CASES[i % 10]}-#{i}", RubyBenchmark.expected(i, user, @generation)] + end + actual = all_flags(user) + RubyBenchmark.require_condition(actual == expected, 'All-flags correctness mismatch') + next unless @spec['mode'] == 'single' + + i = @spec.fetch('flag_index') + RubyBenchmark.require_condition(single_flag(user) == RubyBenchmark.expected(i, user, @generation), + 'Single-flag correctness mismatch') + end + if @spec['mode'] == 'refresh' + RubyBenchmark.require_condition(state.fetch('definitions') == @generation, + 'Reload did not request definitions') + end + else + expected_counts = USERS.each_with_index.to_h do |user, i| + [user[:id], (@captured / 16) + (i < @captured % 16 ? 1 : 0)] + end + valid = state['delivered'] == @captured && state['invalid_events'].zero? && + state['unique_sequences'] == @captured && state['missing_sequences'].zero? && + state['delivered_by_user'] == expected_counts + RubyBenchmark.require_condition(valid, 'Capture payload, identity or sequence mismatch') + end + end + + def prepare(operations) + warmup = if @spec['mode'] == 'refresh' + 2 + else + [operations, @spec['flag_count'] == 1000 ? 32 : 160].min + end + control('reset', { operations: @spec.key?('payload_bytes') ? warmup : 0 }) + prepare_events(warmup) if @spec.key?('payload_bytes') + run(warmup) + flush if @spec['mode'] == 'capture' + verify + reload(0) if @spec.key?('flag_count') + control('reset', { operations: @spec.key?('payload_bytes') ? operations : 0 }) + @captured = 0 + prepare_events(operations) if @spec.key?('payload_bytes') + warmup + end + + def complete + # Enqueue-only verification flush is outside measurement. Delivery verification never rescues a flush. + flush if @spec['mode'] == 'capture' + verify + end + + def close + RubyBenchmark.require_condition(@client.shutdown(timeout: 10), 'Public shutdown timed out') + end + end + + def self.measure(spec, url) + operations = if spec['pass'] == 'allocation' + spec.fetch('allocation_operations', + spec['operations']) + else + spec['operations'] + end + workload = Workload.new(spec, url) + begin + warmup = workload.prepare(operations) + settle + result = { status: 'ok', pass: spec['pass'], operations: operations, warmup_operations: warmup } + case spec['pass'] + when 'timing' + cpu_clock = spec['mode'] == 'capture' ? Process::CLOCK_THREAD_CPUTIME_ID : Process::CLOCK_PROCESS_CPUTIME_ID + cpu = Process.clock_gettime(cpu_clock, :nanosecond) + start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) + workload.run(operations) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) - start + cpu = Process.clock_gettime(cpu_clock, :nanosecond) - cpu + result.merge!(elapsed_ns: elapsed, elapsed_time_per_op: elapsed.to_f / operations, + cpu_time_per_op: cpu.to_f / operations) + when 'allocation' + before = GC.stat(:total_allocated_objects) + workload.run(operations) + result[:allocated_objects] = GC.stat(:total_allocated_objects) - before + when 'memory' + before = snapshot + checkpoints = [] + workload.run(operations) { |i| checkpoints << { operations: i, **snapshot } } + if checkpoints.empty? || checkpoints.last[:operations] != operations + checkpoints << { operations: operations, + **snapshot } + end + after = snapshot + settle + after_settling = snapshot + high_water = if File.file?('/proc/self/status') + File.read('/proc/self/status')[/^VmHWM:\s+(\d+)/, + 1].to_i * 1024 + end + result.merge!(before: before, after: after, after_settling: after_settling, checkpoints: checkpoints, + observed_peak_rss: [before[:rss], after[:rss], *checkpoints.map { |p| p[:rss] }].max, + process_lifetime_peak_rss: high_water) + else + raise 'Unknown measurement pass' + end + workload.complete + result + ensure + workload.close + end + end +end + +if $PROGRAM_NAME == __FILE__ + PostHog::Logging.logger = Logger.new(File::NULL) + if ARGV == ['--describe'] + puts JSON.generate(RubyBenchmark.describe) + else + spec_path, url, output = ARGV + spec = JSON.parse(File.read(spec_path)) + result = RubyBenchmark.measure(spec, url) + File.write(output, JSON.generate(result)) + end +end