ci: integrate shared Python benchmark reports

.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@2a6111d08d784c0ca84f7f97f32a13fdd6d31449 # 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

.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,20 +57,77 @@ 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@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 }}/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()
               id: results
               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
 

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,21 +189,28 @@ 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"],
             }
             for size in (32, 4096)
         ],
         *[
             {
-                "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)
         ],
     ]

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
         }
     )

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,6 +393,9 @@ 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")

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)