From 60f17d6ae7bec5a247999ee92bc9f4d3377a3af8 Mon Sep 17 00:00:00 2001 From: Rayhan Hossain Date: Mon, 10 Aug 2026 15:25:14 -0700 Subject: [PATCH 1/2] Add telemetry benchamark setup Signed-off-by: Rayhan Hossain --- .gitignore | 1 + README.md | 6 + playgrounds/benchmarks/README.md | 12 + .../benchmarks/local-latency/README.md | 99 ++++ .../local-latency/adapters/mongoose.sh | 36 ++ .../local-latency/analyze_jaeger.py | 423 ++++++++++++++++++ .../benchmarks/local-latency/benchmark.sh | 124 +++++ .../schemas/adapter-result.schema.json | 64 +++ .../local-latency/test_analyze_jaeger.py | 113 +++++ playgrounds/mongoose/README.md | 10 + playgrounds/mongoose/app/benchmark.js | 228 ++++++++++ playgrounds/mongoose/app/package.json | 1 + playgrounds/mongoose/app/telemetry.js | 3 +- .../mongoose/scripts/run-telemetry-demo.sh | 2 +- shared/telemetry/README.md | 16 + shared/telemetry/compose.yaml | 2 +- .../telemetry/otel-collector-benchmark.yaml | 35 ++ 17 files changed, 1172 insertions(+), 3 deletions(-) create mode 100644 playgrounds/benchmarks/README.md create mode 100644 playgrounds/benchmarks/local-latency/README.md create mode 100755 playgrounds/benchmarks/local-latency/adapters/mongoose.sh create mode 100755 playgrounds/benchmarks/local-latency/analyze_jaeger.py create mode 100755 playgrounds/benchmarks/local-latency/benchmark.sh create mode 100644 playgrounds/benchmarks/local-latency/schemas/adapter-result.schema.json create mode 100644 playgrounds/benchmarks/local-latency/test_analyze_jaeger.py create mode 100644 playgrounds/mongoose/app/benchmark.js create mode 100644 shared/telemetry/otel-collector-benchmark.yaml diff --git a/.gitignore b/.gitignore index 7415c77..e281a80 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,7 @@ __pycache__/ # Runtime / logs *.log logs/ +playgrounds/benchmarks/*/results/ # Environment / secrets .env diff --git a/README.md b/README.md index e7bbb0f..9e1dd43 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,7 @@ playground's README states what it needs. | [mongodb-node](playgrounds/mongodb-node/) | Node.js — MongoDB driver | Express REST API + a CRUD/compatibility suite using the native Node.js driver. | | [beanie](playgrounds/beanie/) | Python — Beanie ODM | FastAPI REST API + a CRUD/compatibility test suite using the Beanie ODM. | | [pymongo](playgrounds/pymongo/) | Python — PyMongo driver | Flask REST API + a CRUD/compatibility test suite using the raw PyMongo driver. | +| [benchmarks](playgrounds/benchmarks/) | Shared + adapters | Repeatable performance experiments, including local latency analysis. | More MongoDB driver playgrounds are planned. Contributions are welcome. @@ -80,6 +81,10 @@ cd playgrounds/mongoose See the [shared telemetry guide](shared/telemetry/README.md) for image, configuration, and verification details. +For configurable baseline and traced load experiments, use the +[local latency benchmark](playgrounds/benchmarks/local-latency/). It currently +includes a Mongoose adapter and a common contract for additional drivers. + ## Repository Layout ``` @@ -89,6 +94,7 @@ documentdb-playground/ ├── shared/ │ └── telemetry/ # Collector + Jaeger + tracing-enabled DocumentDB stack └── playgrounds/ + ├── benchmarks/ # Performance experiments + driver adapters ├── mongoose/ # Node.js + Mongoose ODM ├── mongodb-node/ # Node.js + MongoDB native driver ├── beanie/ # Python + Beanie ODM diff --git a/playgrounds/benchmarks/README.md b/playgrounds/benchmarks/README.md new file mode 100644 index 0000000..0524c64 --- /dev/null +++ b/playgrounds/benchmarks/README.md @@ -0,0 +1,12 @@ +# DocumentDB benchmarks + +This directory contains repeatable synthetic performance experiments. Each +benchmark documents its workload, controls, measurement boundaries, artifacts, +and interpretation limits. Results are diagnostic and are not production +capacity claims. + +## Available benchmarks + +| Benchmark | Purpose | +| --- | --- | +| [Local latency](local-latency/) | Compare untraced and traced client latency, then decompose connected traces into client, gateway, and PostgreSQL segments. | \ No newline at end of file diff --git a/playgrounds/benchmarks/local-latency/README.md b/playgrounds/benchmarks/local-latency/README.md new file mode 100644 index 0000000..3cc3bc7 --- /dev/null +++ b/playgrounds/benchmarks/local-latency/README.md @@ -0,0 +1,99 @@ +# DocumentDB local latency benchmark + +This playground runs the same driver-native workload twice: once without +application tracing and once with a connected application-to-gateway trace. It +stores client percentiles, throughput, trace-segment percentiles, and direct +Jaeger links in a durable experiment directory. + +The initial adapter uses Mongoose. Shared orchestration invokes adapters as +processes and does not import Node.js or Python application code. + +## Workload data + +The `large-read` workload creates one book before measurement: + +```json +{ + "title": "documentdb-benchmark-target", + "author": "benchmark:", + "genres": ["benchmark"], + "pages": 1 +} +``` + +It builds a deterministic `$in` query containing that title followed by +nonmatching strings until the driver's BSON serializer reports at least +`BENCHMARK_QUERY_BYTES`. Seeding, index initialization, cleanup, and warm-up are +outside the measured interval. No production or external data is ingested. + +## Concurrency + +`BENCHMARK_CONCURRENCY` is the maximum number of in-flight database operations. +Each adapter uses a bounded worker pool: a worker starts its next operation only +after its previous operation finishes. Client latency uses a monotonic clock. + +## Run + +```bash +BENCHMARK_OPERATIONS=100 \ +BENCHMARK_WARMUP=20 \ +BENCHMARK_CONCURRENCY=4 \ +BENCHMARK_QUERY_BYTES=4096 \ + ./playgrounds/benchmarks/local-latency/benchmark.sh +``` + +For a higher-volume experiment, increase operations, warm-up, concurrency, and +query size. Repeat longer runs before drawing conclusions from tail percentiles. + +| Variable | Default | Meaning | +| --- | --- | --- | +| `BENCHMARK_ADAPTER` | `mongoose` | Adapter under `adapters/`. | +| `BENCHMARK_ID` | generated | Experiment and trace correlation ID. | +| `BENCHMARK_OPERATIONS` | `1000` | Measured operations per phase. | +| `BENCHMARK_WARMUP` | `50` | Unmeasured operations per phase. | +| `BENCHMARK_CONCURRENCY` | `10` | Maximum in-flight operations. | +| `BENCHMARK_QUERY_BYTES` | `16384` | Minimum serialized BSON query size. | +| `BENCHMARK_WORKLOAD` | `large-read` | Portable workload name. | +| `BENCHMARK_RESULTS_DIR` | `results/` | Artifact parent directory. | +| `DOCUMENTDB_IMAGE` | `ghcr.io/documentdb/documentdb/documentdb-local:trace-4fbbfcb8` | Official tracing-enabled image. | + +## Artifacts and Jaeger + +Every run creates an ignored `results//` directory: + +| File | Contents | +| --- | --- | +| `experiment.json` | Image identity and artifact manifest. | +| `summary.md` | Human-readable client and hop summary with measurement limitations. | +| `baseline.json` | Client results with application tracing disabled. | +| `traced.json` | Client results with application tracing enabled. | +| `trace-analysis.json` | Jaeger counts, exclusions, percentiles, and links. | +| `trace-segments.csv` | One row per complete connected trace. | + +The analysis includes client total, client time outside the gateway, gateway +total, PostgreSQL duration sum and interval union, and gateway residual. The +residual subtracts the interval union so overlapping database spans are not +double-counted. Open the median and p99 links and compare the client span with +`gateway.request`, `gateway.process_request`, and `postgres.execute`. + +Cross-process clocks can be slightly skewed, so the analyzer does not claim to +split request and response transport accurately. `client_outside_gateway` +includes driver work, pool wait, serialization, TLS, network time, and response +decoding. Nested duration subtraction is not CPU profiling. Jaeger storage is +ephemeral; retain the JSON and CSV artifacts after stopping the stack. + +The benchmark Collector profile omits the verbose debug exporter. Leave SQL +commenter disabled unless SQL-log correlation is the experiment. + +## Add an adapter + +Add executable `adapters/.sh` plus driver-native workload code in that +driver's playground. The adapter receives the `BENCHMARK_*` variables and must +atomically write `BENCHMARK_RESULT_FILE` according to +`schemas/adapter-result.schema.json`. + +For each measured traced operation, emit a client span with `benchmark.id`, +`benchmark.phase`, `benchmark.query_bytes`, `benchmark.concurrency`, +`benchmark.workload`, and `benchmark.adapter`. Inject W3C `traceparent` into the +MongoDB command `comment` so `gateway.request` becomes its child. Reusing the +shared containers alone does not provide driver-side instrumentation. \ No newline at end of file diff --git a/playgrounds/benchmarks/local-latency/adapters/mongoose.sh b/playgrounds/benchmarks/local-latency/adapters/mongoose.sh new file mode 100755 index 0000000..c57f91f --- /dev/null +++ b/playgrounds/benchmarks/local-latency/adapters/mongoose.sh @@ -0,0 +1,36 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +PLAYGROUND_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" +MONGOOSE_DIR="$PLAYGROUND_ROOT/mongoose" + +for command_name in node npm; do + command -v "$command_name" >/dev/null || { + echo "$command_name is required by the Mongoose benchmark adapter" >&2 + exit 1 + } +done + +for variable_name in \ + BENCHMARK_ID \ + BENCHMARK_PHASE \ + BENCHMARK_RESULT_FILE; do + if [ -z "${!variable_name:-}" ]; then + echo "$variable_name is required by the Mongoose benchmark adapter" >&2 + exit 1 + fi +done + +# shellcheck source=../../mongoose/scripts/lib.sh +source "$MONGOOSE_DIR/scripts/lib.sh" + +if [ -z "${MONGO_URI:-}" ]; then + MONGO_URI="$(build_uri)" + export MONGO_URI +fi +export MONGO_DB="${MONGO_DB:-mongoose_benchmark}" + +(cd "$MONGOOSE_DIR/app" && npm install --omit=dev --no-audit --no-fund >/dev/null) +exec node "$MONGOOSE_DIR/app/benchmark.js" \ No newline at end of file diff --git a/playgrounds/benchmarks/local-latency/analyze_jaeger.py b/playgrounds/benchmarks/local-latency/analyze_jaeger.py new file mode 100755 index 0000000..2ee571a --- /dev/null +++ b/playgrounds/benchmarks/local-latency/analyze_jaeger.py @@ -0,0 +1,423 @@ +#!/usr/bin/env python3 + +import argparse +import csv +import json +import math +import os +import sys +import tempfile +import time +import urllib.error +import urllib.parse +import urllib.request + + +def percentile(values, percentage): + ordered = sorted(values) + if not ordered: + return None + index = max(0, math.ceil(percentage / 100 * len(ordered)) - 1) + return ordered[index] + + +def summarize(values): + if not values: + return None + return { + "count": len(values), + "minimum": min(values), + "mean": sum(values) / len(values), + "p50": percentile(values, 50), + "p95": percentile(values, 95), + "p99": percentile(values, 99), + "maximum": max(values), + } + + +def interval_union_duration(intervals): + if not intervals: + return 0 + ordered = sorted(intervals) + start, end = ordered[0] + total = 0 + for next_start, next_end in ordered[1:]: + if next_start <= end: + end = max(end, next_end) + else: + total += end - start + start, end = next_start, next_end + return total + end - start + + +def tags(span): + return {tag.get("key"): tag.get("value") for tag in span.get("tags", [])} + + +def child_of(span, parent_span_id): + return any( + reference.get("refType") == "CHILD_OF" + and reference.get("spanID") == parent_span_id + for reference in span.get("references", []) + ) + + +def complete_trace_count(traces, result, service_name, gateway_service): + count = 0 + for trace in traces: + processes = trace.get("processes", {}) + + def service(span): + return processes.get(span.get("processID"), {}).get("serviceName") + + spans = trace.get("spans", []) + application = next( + ( + span + for span in spans + if service(span) == service_name + and tags(span).get("benchmark.id") == result["benchmarkId"] + and tags(span).get("benchmark.phase") == "traced" + and result["measurementStartUs"] + <= int(span.get("startTime", 0)) + <= result["measurementEndUs"] + ), + None, + ) + if application is None: + continue + gateway = next( + ( + span + for span in spans + if service(span) == gateway_service + and span.get("operationName") == "gateway.request" + and child_of(span, application.get("spanID")) + ), + None, + ) + if gateway is None: + continue + if any( + service(span) == gateway_service + and span.get("operationName") == "postgres.execute" + for span in spans + ): + count += 1 + return count + + +def write_json_atomic(file_name, value): + directory = os.path.dirname(os.path.abspath(file_name)) + os.makedirs(directory, exist_ok=True) + descriptor, temporary = tempfile.mkstemp(prefix=".analysis-", dir=directory) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(value, handle, indent=2) + handle.write("\n") + os.replace(temporary, file_name) + except Exception: + os.unlink(temporary) + raise + + +def write_text_atomic(file_name, value): + directory = os.path.dirname(os.path.abspath(file_name)) + os.makedirs(directory, exist_ok=True) + descriptor, temporary = tempfile.mkstemp(prefix=".summary-", dir=directory) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(value) + os.replace(temporary, file_name) + except Exception: + os.unlink(temporary) + raise + + +def build_markdown_summary(baseline, traced, analysis): + def client_row(label, value): + latency = value["latencyMs"] + return ( + f"| {label} | {value['throughputOperationsPerSecond']:.1f} | " + f"{latency['mean']:.3f} | {latency['p50']:.3f} | " + f"{latency['p95']:.3f} | {latency['p99']:.3f} | " + f"{latency['maximum']:.3f} |" + ) + + segment_labels = { + "client_total_ms": "Full client operation", + "client_outside_gateway_ms": "Client/driver/transport outside gateway", + "gateway_total_ms": "Gateway total", + "postgres_union_ms": "PostgreSQL interval union", + "gateway_residual_ms": "Gateway excluding PostgreSQL", + } + hop_rows = [] + for segment_name, label in segment_labels.items(): + segment = analysis["segments"][segment_name] + hop_rows.append( + f"| {label} | {segment['mean']:.3f} | {segment['p50']:.3f} | " + f"{segment['p95']:.3f} | {segment['p99']:.3f} | " + f"{segment['maximum']:.3f} |" + ) + + traces = analysis["representativeTraces"] + return "\n".join( + [ + f"# Benchmark summary: {traced['benchmarkId']}", + "", + f"- Adapter: `{traced['adapter']}`", + f"- Workload: `{traced['workload']}`", + f"- Operations per phase: `{traced['measuredOperations']}`", + f"- Warm-up operations per phase: `{traced['warmupOperations']}`", + f"- Concurrency: `{traced['concurrency']}`", + f"- Query BSON size: `{traced['actualQueryBytes']}` bytes", + f"- Connected traces: `{analysis['completeTraceCount']}/{analysis['returnedTraceCount']}`", + "", + "> [!IMPORTANT]", + "> Exact one-way client-to-gateway latency is not available because the client and gateway use different wall clocks. `Client/driver/transport outside gateway` is clock-robust, but combines client processing, pool wait, serialization, TLS, both network directions, and response decoding.", + "", + "## Client results", + "", + "| Phase | Throughput (ops/s) | Mean (ms) | p50 (ms) | p95 (ms) | p99 (ms) | Max (ms) |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: |", + client_row("Baseline", baseline), + client_row("Traced", traced), + "", + "## Latency by hop", + "", + "| Segment | Mean (ms) | p50 (ms) | p95 (ms) | p99 (ms) | Max (ms) |", + "| --- | ---: | ---: | ---: | ---: | ---: |", + *hop_rows, + "", + "## Representative traces", + "", + f"- [Median trace]({traces['median']})", + f"- [p99 trace]({traces['p99']})", + "", + ] + ) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--baseline", required=True, help="Baseline adapter result JSON") + parser.add_argument("--result", required=True, help="Traced adapter result JSON") + parser.add_argument("--output", required=True, help="Analysis JSON output") + parser.add_argument("--csv", required=True, help="Per-trace CSV output") + parser.add_argument("--summary", required=True, help="Markdown summary output") + parser.add_argument("--jaeger-url", default="http://localhost:16686") + parser.add_argument("--gateway-service", default="documentdb_gateway") + parser.add_argument("--search-attempts", type=int, default=15) + parser.add_argument("--search-delay", type=float, default=1.0) + args = parser.parse_args() + + with open(args.baseline, encoding="utf-8") as handle: + baseline = json.load(handle) + with open(args.result, encoding="utf-8") as handle: + result = json.load(handle) + service_name = result.get("telemetry", {}).get("serviceName") + if not service_name: + raise SystemExit("Adapter result does not contain telemetry.serviceName") + + query = urllib.parse.urlencode( + { + "service": service_name, + "start": result["measurementStartUs"], + "end": result["measurementEndUs"] + 5_000_000, + "limit": max(100, result["measuredOperations"] * 2), + } + ) + traces = [] + for attempt in range(args.search_attempts): + try: + with urllib.request.urlopen( + f"{args.jaeger_url}/api/traces?{query}", timeout=15 + ) as response: + traces = json.load(response).get("data", []) + except (urllib.error.URLError, TimeoutError) as error: + if attempt == args.search_attempts - 1: + raise SystemExit( + f"Could not query Jaeger at {args.jaeger_url}: {error}" + ) + if complete_trace_count( + traces, result, service_name, args.gateway_service + ) >= result["successfulOperations"]: + break + if attempt < args.search_attempts - 1: + time.sleep(args.search_delay) + + excluded = { + "missing_application_span": 0, + "outside_measurement_window": 0, + "missing_gateway_span": 0, + "missing_postgres_span": 0, + "unexpected_parent_relationship": 0, + "negative_derived_duration": 0, + } + rows = [] + response_boundary_clock_skew = [] + for trace in traces: + processes = trace.get("processes", {}) + + def service(span): + return processes.get(span.get("processID"), {}).get("serviceName") + + spans = trace.get("spans", []) + application = next( + ( + span + for span in spans + if service(span) == service_name + and tags(span).get("benchmark.id") == result["benchmarkId"] + and tags(span).get("benchmark.phase") == "traced" + ), + None, + ) + if application is None: + excluded["missing_application_span"] += 1 + continue + if not ( + result["measurementStartUs"] + <= int(application.get("startTime", 0)) + <= result["measurementEndUs"] + ): + excluded["outside_measurement_window"] += 1 + continue + gateway = next( + ( + span + for span in spans + if service(span) == args.gateway_service + and span.get("operationName") == "gateway.request" + ), + None, + ) + if gateway is None: + excluded["missing_gateway_span"] += 1 + continue + if not child_of(gateway, application.get("spanID")): + excluded["unexpected_parent_relationship"] += 1 + continue + postgres = [ + span + for span in spans + if service(span) == args.gateway_service + and span.get("operationName") == "postgres.execute" + ] + if not postgres: + excluded["missing_postgres_span"] += 1 + continue + + client_total = int(application["duration"]) + gateway_total = int(gateway["duration"]) + postgres_sum = sum(int(span["duration"]) for span in postgres) + postgres_union = interval_union_duration( + [ + ( + int(span["startTime"]), + int(span["startTime"]) + int(span["duration"]), + ) + for span in postgres + ] + ) + row = { + "trace_id": trace.get("traceID"), + "client_total_ms": client_total / 1000, + "client_outside_gateway_ms": (client_total - gateway_total) / 1000, + "gateway_total_ms": gateway_total / 1000, + "postgres_sum_ms": postgres_sum / 1000, + "postgres_union_ms": postgres_union / 1000, + "gateway_residual_ms": (gateway_total - postgres_union) / 1000, + } + if any(value < 0 for key, value in row.items() if key.endswith("_ms")): + excluded["negative_derived_duration"] += 1 + continue + + response_return = ( + int(application["startTime"]) + + client_total + - int(gateway["startTime"]) + - gateway_total + ) + if response_return < 0: + response_boundary_clock_skew.append(response_return) + rows.append(row) + + if not rows: + raise SystemExit( + "No complete connected benchmark traces found; " + f"excluded={json.dumps(excluded, sort_keys=True)}" + ) + if len(rows) < result["successfulOperations"]: + raise SystemExit( + "Incomplete connected benchmark traces after waiting for Jaeger; " + f"expected={result['successfulOperations']} complete={len(rows)} " + f"excluded={json.dumps(excluded, sort_keys=True)}" + ) + + segment_names = [ + "client_total_ms", + "client_outside_gateway_ms", + "gateway_total_ms", + "postgres_sum_ms", + "postgres_union_ms", + "gateway_residual_ms", + ] + summaries = { + name: summarize([row[name] for row in rows]) for name in segment_names + } + ordered = sorted(rows, key=lambda row: row["client_total_ms"]) + median_row = ordered[max(0, math.ceil(0.50 * len(ordered)) - 1)] + p99_row = ordered[max(0, math.ceil(0.99 * len(ordered)) - 1)] + analysis = { + "schemaVersion": 1, + "benchmarkId": result["benchmarkId"], + "adapter": result["adapter"], + "jaegerUrl": args.jaeger_url, + "returnedTraceCount": len(traces), + "completeTraceCount": len(rows), + "excluded": excluded, + "clockSkew": { + "negativeResponseBoundaryCount": len(response_boundary_clock_skew), + "minimumResponseBoundaryUs": min(response_boundary_clock_skew, default=0), + "note": "Cross-process timestamps cannot reliably split request and response transport. Use client_outside_gateway_ms.", + }, + "measurementScope": { + "oneWayClientToGatewayAvailable": False, + "clientOutsideGatewayIncludes": [ + "client processing", + "connection pool wait", + "serialization", + "TLS", + "client-to-gateway transport", + "gateway-to-client transport", + "response decoding", + ], + "note": "Client and gateway wall clocks are not synchronized; use client_outside_gateway_ms as the clock-robust combined measurement.", + }, + "segments": summaries, + "representativeTraces": { + "median": f"{args.jaeger_url}/trace/{median_row['trace_id']}", + "p99": f"{args.jaeger_url}/trace/{p99_row['trace_id']}", + }, + } + write_json_atomic(args.output, analysis) + write_text_atomic(args.summary, build_markdown_summary(baseline, result, analysis)) + with open(args.csv, "w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + + print(f"Connected traces: {len(rows)}/{len(traces)}") + print("segment\tp50_ms\tp95_ms\tp99_ms") + for name in segment_names: + summary = summaries[name] + print( + f"{name}\t{summary['p50']:.3f}\t{summary['p95']:.3f}\t{summary['p99']:.3f}" + ) + print(f"Median trace: {analysis['representativeTraces']['median']}") + print(f"P99 trace: {analysis['representativeTraces']['p99']}") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/playgrounds/benchmarks/local-latency/benchmark.sh b/playgrounds/benchmarks/local-latency/benchmark.sh new file mode 100755 index 0000000..01a8195 --- /dev/null +++ b/playgrounds/benchmarks/local-latency/benchmark.sh @@ -0,0 +1,124 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" +TELEMETRY_DIR="$REPO_ROOT/shared/telemetry" +ADAPTER="${BENCHMARK_ADAPTER:-mongoose}" +ADAPTER_SCRIPT="$SCRIPT_DIR/adapters/$ADAPTER.sh" +BENCHMARK_ID="${BENCHMARK_ID:-$(date -u +%Y%m%dT%H%M%SZ)-$ADAPTER-$$}" +RESULTS_ROOT="${BENCHMARK_RESULTS_DIR:-$SCRIPT_DIR/results}" +RESULT_DIR="$RESULTS_ROOT/$BENCHMARK_ID" +BASELINE_RESULT="$RESULT_DIR/baseline.json" +TRACED_RESULT="$RESULT_DIR/traced.json" +ANALYSIS_RESULT="$RESULT_DIR/trace-analysis.json" +TRACE_CSV="$RESULT_DIR/trace-segments.csv" +SUMMARY_RESULT="$RESULT_DIR/summary.md" +EXPERIMENT_RESULT="$RESULT_DIR/experiment.json" + +if [ ! -x "$ADAPTER_SCRIPT" ]; then + echo "Unknown or non-executable benchmark adapter: $ADAPTER" >&2 + echo "Expected: $ADAPTER_SCRIPT" >&2 + exit 1 +fi +for command_name in docker curl python3; do + command -v "$command_name" >/dev/null || { + echo "$command_name is required" >&2 + exit 1 + } +done +if [ -e "$RESULT_DIR" ]; then + echo "Benchmark result directory already exists: $RESULT_DIR" >&2 + exit 1 +fi +mkdir -p "$RESULT_DIR" + +export BENCHMARK_ID +export BENCHMARK_OPERATIONS="${BENCHMARK_OPERATIONS:-1000}" +export BENCHMARK_WARMUP="${BENCHMARK_WARMUP:-50}" +export BENCHMARK_CONCURRENCY="${BENCHMARK_CONCURRENCY:-10}" +export BENCHMARK_QUERY_BYTES="${BENCHMARK_QUERY_BYTES:-16384}" +export BENCHMARK_WORKLOAD="${BENCHMARK_WORKLOAD:-large-read}" +export DOCUMENTDB_IMAGE="${DOCUMENTDB_IMAGE:-ghcr.io/documentdb/documentdb/documentdb-local:trace-4fbbfcb8}" +export OTEL_COLLECTOR_CONFIG_FILE="${OTEL_COLLECTOR_CONFIG_FILE:-./otel-collector-benchmark.yaml}" +export OTEL_EXPORTER_OTLP_ENDPOINT="${OTEL_EXPORTER_OTLP_ENDPOINT:-http://localhost:${OTEL_COLLECTOR_GRPC_PORT:-4317}}" +export OTEL_SERVICE_NAME="${OTEL_SERVICE_NAME:-documentdb-benchmark-$ADAPTER}" + +echo "Starting benchmark telemetry stack with $DOCUMENTDB_IMAGE" +"$TELEMETRY_DIR/scripts/up.sh" + +run_phase() { + local phase="$1" + local tracing_enabled="$2" + local result_file="$3" + echo "Running $phase phase: operations=$BENCHMARK_OPERATIONS warmup=$BENCHMARK_WARMUP concurrency=$BENCHMARK_CONCURRENCY query_bytes=$BENCHMARK_QUERY_BYTES" + BENCHMARK_PHASE="$phase" \ + BENCHMARK_RESULT_FILE="$result_file" \ + OTEL_TRACES_ENABLED="$tracing_enabled" \ + "$ADAPTER_SCRIPT" +} + +run_phase baseline false "$BASELINE_RESULT" +run_phase traced true "$TRACED_RESULT" + +python3 "$SCRIPT_DIR/analyze_jaeger.py" \ + --baseline "$BASELINE_RESULT" \ + --result "$TRACED_RESULT" \ + --output "$ANALYSIS_RESULT" \ + --csv "$TRACE_CSV" \ + --summary "$SUMMARY_RESULT" \ + --jaeger-url "${JAEGER_URL:-http://localhost:${JAEGER_UI_PORT:-16686}}" + +IMAGE_ID="$(docker image inspect "$DOCUMENTDB_IMAGE" --format '{{.Id}}')" +IMAGE_CREATED="$(docker image inspect "$DOCUMENTDB_IMAGE" --format '{{.Created}}')" +python3 - \ + "$EXPERIMENT_RESULT" \ + "$BENCHMARK_ID" \ + "$ADAPTER" \ + "$DOCUMENTDB_IMAGE" \ + "$IMAGE_ID" \ + "$IMAGE_CREATED" \ + "$BASELINE_RESULT" \ + "$TRACED_RESULT" \ + "$ANALYSIS_RESULT" \ + "$TRACE_CSV" \ + "$SUMMARY_RESULT" <<'PY' +import json +import os +import sys +import tempfile + +output, benchmark_id, adapter, image, image_id, image_created, baseline, traced, analysis, trace_csv, summary = sys.argv[1:] +value = { + "schemaVersion": 1, + "benchmarkId": benchmark_id, + "adapter": adapter, + "documentdbImage": { + "name": image, + "id": image_id, + "created": image_created, + }, + "artifacts": { + "baseline": os.path.basename(baseline), + "traced": os.path.basename(traced), + "traceAnalysis": os.path.basename(analysis), + "traceSegments": os.path.basename(trace_csv), + "summary": os.path.basename(summary), + }, +} +directory = os.path.dirname(output) +descriptor, temporary = tempfile.mkstemp(prefix=".experiment-", dir=directory) +with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(value, handle, indent=2) + handle.write("\n") +os.replace(temporary, output) +PY + +echo "" +echo "Benchmark complete: $RESULT_DIR" +echo " Baseline: $BASELINE_RESULT" +echo " Traced: $TRACED_RESULT" +echo " Analysis: $ANALYSIS_RESULT" +echo " CSV: $TRACE_CSV" +echo " Summary: $SUMMARY_RESULT" \ No newline at end of file diff --git a/playgrounds/benchmarks/local-latency/schemas/adapter-result.schema.json b/playgrounds/benchmarks/local-latency/schemas/adapter-result.schema.json new file mode 100644 index 0000000..2454e40 --- /dev/null +++ b/playgrounds/benchmarks/local-latency/schemas/adapter-result.schema.json @@ -0,0 +1,64 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://documentdb.io/schemas/benchmark-adapter-result-v1.json", + "title": "DocumentDB benchmark adapter result", + "type": "object", + "required": [ + "schemaVersion", + "adapter", + "benchmarkId", + "phase", + "workload", + "tracingEnabled", + "requestedQueryBytes", + "actualQueryBytes", + "warmupOperations", + "measuredOperations", + "successfulOperations", + "failedOperations", + "concurrency", + "measurementStartUs", + "measurementEndUs", + "elapsedSeconds", + "throughputOperationsPerSecond", + "latencyMs", + "runtime", + "telemetry", + "database" + ], + "properties": { + "schemaVersion": { "const": 1 }, + "adapter": { "type": "string", "minLength": 1 }, + "benchmarkId": { "type": "string", "minLength": 1 }, + "phase": { "enum": ["baseline", "traced"] }, + "workload": { "type": "string", "minLength": 1 }, + "tracingEnabled": { "type": "boolean" }, + "requestedQueryBytes": { "type": "integer", "minimum": 1 }, + "actualQueryBytes": { "type": "integer", "minimum": 1 }, + "warmupOperations": { "type": "integer", "minimum": 0 }, + "measuredOperations": { "type": "integer", "minimum": 1 }, + "successfulOperations": { "type": "integer", "minimum": 0 }, + "failedOperations": { "type": "integer", "minimum": 0 }, + "concurrency": { "type": "integer", "minimum": 1 }, + "measurementStartUs": { "type": "integer", "minimum": 1 }, + "measurementEndUs": { "type": "integer", "minimum": 1 }, + "elapsedSeconds": { "type": "number", "exclusiveMinimum": 0 }, + "throughputOperationsPerSecond": { "type": "number", "minimum": 0 }, + "latencyMs": { + "type": "object", + "required": ["minimum", "mean", "p50", "p95", "p99", "maximum"], + "additionalProperties": { "type": "number", "minimum": 0 } + }, + "runtime": { "type": "object" }, + "telemetry": { + "type": "object", + "required": ["serviceName", "operationName"], + "properties": { + "serviceName": { "type": ["string", "null"] }, + "operationName": { "type": "string" } + } + }, + "database": { "type": "object" } + }, + "additionalProperties": true +} \ No newline at end of file diff --git a/playgrounds/benchmarks/local-latency/test_analyze_jaeger.py b/playgrounds/benchmarks/local-latency/test_analyze_jaeger.py new file mode 100644 index 0000000..97c4f83 --- /dev/null +++ b/playgrounds/benchmarks/local-latency/test_analyze_jaeger.py @@ -0,0 +1,113 @@ +import unittest + +from analyze_jaeger import ( + build_markdown_summary, + complete_trace_count, + interval_union_duration, + percentile, +) + + +class AnalyzeJaegerTest(unittest.TestCase): + def test_percentile_uses_nearest_rank(self): + self.assertEqual(percentile([4, 1, 3, 2], 50), 2) + self.assertEqual(percentile([4, 1, 3, 2], 99), 4) + + def test_interval_union_merges_overlaps_and_adjacency(self): + self.assertEqual( + interval_union_duration([(20, 25), (1, 5), (4, 10), (10, 12)]), + 16, + ) + + def test_interval_union_handles_empty_input(self): + self.assertEqual(interval_union_duration([]), 0) + + def test_summary_discloses_one_way_latency_limit(self): + result = { + "benchmarkId": "test-run", + "adapter": "test", + "workload": "large-read", + "measuredOperations": 10, + "warmupOperations": 2, + "concurrency": 1, + "actualQueryBytes": 1024, + "throughputOperationsPerSecond": 100, + "latencyMs": { + "mean": 1, + "p50": 1, + "p95": 2, + "p99": 3, + "maximum": 4, + }, + } + segment = {"mean": 1, "p50": 1, "p95": 2, "p99": 3, "maximum": 4} + analysis = { + "completeTraceCount": 10, + "returnedTraceCount": 10, + "segments": { + "client_total_ms": segment, + "client_outside_gateway_ms": segment, + "gateway_total_ms": segment, + "postgres_union_ms": segment, + "gateway_residual_ms": segment, + }, + "representativeTraces": {"median": "http://median", "p99": "http://p99"}, + } + + summary = build_markdown_summary(result, result, analysis) + + self.assertIn("Exact one-way client-to-gateway latency is not available", summary) + self.assertIn("## Latency by hop", summary) + + def test_complete_trace_count_requires_connected_postgres_span(self): + result = { + "benchmarkId": "test-run", + "measurementStartUs": 100, + "measurementEndUs": 200, + } + trace = { + "processes": { + "app": {"serviceName": "client"}, + "gateway": {"serviceName": "documentdb_gateway"}, + }, + "spans": [ + { + "processID": "app", + "spanID": "client-span", + "startTime": 150, + "tags": [ + {"key": "benchmark.id", "value": "test-run"}, + {"key": "benchmark.phase", "value": "traced"}, + ], + }, + { + "processID": "gateway", + "operationName": "gateway.request", + "references": [ + {"refType": "CHILD_OF", "spanID": "client-span"} + ], + }, + { + "processID": "gateway", + "operationName": "postgres.execute", + }, + ], + } + + self.assertEqual( + complete_trace_count( + [trace], result, "client", "documentdb_gateway" + ), + 1, + ) + trace["spans"].pop() + self.assertEqual( + complete_trace_count( + [trace], result, "client", "documentdb_gateway" + ), + 0, + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/playgrounds/mongoose/README.md b/playgrounds/mongoose/README.md index 5d61eee..543626e 100644 --- a/playgrounds/mongoose/README.md +++ b/playgrounds/mongoose/README.md @@ -233,6 +233,16 @@ Read by [`app/db.js`](app/db.js), [`app/server.js`](app/server.js), and | `OTEL_EXPORTER_OTLP_ENDPOINT` | OpenTelemetry default | app | Collector endpoint used by the traced app. | | `OTEL_SERVICE_NAME` | SDK default | app | Application service name shown in Jaeger. | +The telemetry demo and benchmark use the official tracing-enabled image by +default: + +```text +ghcr.io/documentdb/documentdb/documentdb-local:trace-4fbbfcb8 +``` + +This does not change the regular app and CRUD scripts, which continue to use +the standard `latest` image unless `DOCUMENTDB_IMAGE` is set. + ## Running the Test Suite Manually The scripts handle everything, but you can also run the suite directly against diff --git a/playgrounds/mongoose/app/benchmark.js b/playgrounds/mongoose/app/benchmark.js new file mode 100644 index 0000000..b70c338 --- /dev/null +++ b/playgrounds/mongoose/app/benchmark.js @@ -0,0 +1,228 @@ +'use strict'; + +const fs = require('fs'); +const os = require('os'); +const path = require('path'); + +const { connect } = require('./db'); +const { + installMongooseTracePropagation, + shutdownTelemetry, + traceDocumentDbOperation, + tracingEnabled, +} = require('./telemetry'); +const mongoose = require('mongoose'); +const Book = require('./models/book'); + +const adapter = 'mongoose'; +const benchmarkId = requiredString('BENCHMARK_ID'); +const phase = requiredChoice('BENCHMARK_PHASE', ['baseline', 'traced']); +const workload = process.env.BENCHMARK_WORKLOAD || 'large-read'; +const resultFile = path.resolve(requiredString('BENCHMARK_RESULT_FILE')); +const operations = positiveInteger('BENCHMARK_OPERATIONS', 1000); +const warmup = nonnegativeInteger('BENCHMARK_WARMUP', 50); +const concurrency = positiveInteger('BENCHMARK_CONCURRENCY', 10); +const requestedQueryBytes = positiveInteger('BENCHMARK_QUERY_BYTES', 16384); +const databaseName = process.env.MONGO_DB || 'mongoose_benchmark'; + +if (workload !== 'large-read') { + throw new Error(`Unsupported Mongoose benchmark workload: ${workload}`); +} +if (phase === 'traced' && !tracingEnabled) { + throw new Error('BENCHMARK_PHASE=traced requires OTEL_TRACES_ENABLED=true'); +} + +function requiredString(name) { + const value = process.env[name]; + if (!value) throw new Error(`${name} is required`); + return value; +} + +function requiredChoice(name, choices) { + const value = requiredString(name); + if (!choices.includes(value)) { + throw new Error(`${name} must be one of: ${choices.join(', ')}`); + } + return value; +} + +function integer(name, fallback, minimum) { + const raw = process.env[name] || String(fallback); + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < minimum) { + throw new Error(`${name} must be an integer greater than or equal to ${minimum}`); + } + return value; +} + +function positiveInteger(name, fallback) { + return integer(name, fallback, 1); +} + +function nonnegativeInteger(name, fallback) { + return integer(name, fallback, 0); +} + +function percentile(sortedValues, percentage) { + if (sortedValues.length === 0) return null; + const index = Math.max(0, Math.ceil((percentage / 100) * sortedValues.length) - 1); + return sortedValues[index]; +} + +function buildLargeReadFilter(targetBytes, matchingTitle) { + const titles = [matchingTitle]; + let filter = { title: { $in: titles } }; + while (mongoose.mongo.BSON.calculateObjectSize(filter) < targetBytes) { + titles.push(`missing-${String(titles.length).padStart(6, '0')}-${'x'.repeat(32)}`); + filter = { title: { $in: titles } }; + } + return { + filter, + actualBytes: mongoose.mongo.BSON.calculateObjectSize(filter), + }; +} + +async function runBounded(total, workerCount, operation) { + let nextIndex = 0; + await Promise.all( + Array.from({ length: Math.min(total, workerCount) }, async () => { + while (true) { + const index = nextIndex; + nextIndex += 1; + if (index >= total) return; + await operation(index); + } + }) + ); +} + +function writeJsonAtomic(file, value) { + fs.mkdirSync(path.dirname(file), { recursive: true }); + const temporary = `${file}.tmp-${process.pid}`; + fs.writeFileSync(temporary, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 }); + fs.renameSync(temporary, file); +} + +async function main() { + installMongooseTracePropagation(mongoose); + await connect(); + await Book.init(); + + const seedAuthor = `benchmark:${benchmarkId}`; + await Book.deleteMany({ author: seedAuthor }); + const seeded = await Book.create({ + title: 'documentdb-benchmark-target', + author: seedAuthor, + genres: ['benchmark'], + pages: 1, + }); + const { filter, actualBytes } = buildLargeReadFilter(requestedQueryBytes, seeded.title); + const attributes = { + 'benchmark.id': benchmarkId, + 'benchmark.phase': phase, + 'benchmark.query_bytes': actualBytes, + 'benchmark.concurrency': concurrency, + 'benchmark.workload': workload, + 'benchmark.adapter': adapter, + }; + const runOperation = () => + traceDocumentDbOperation( + 'benchmarkFind', + databaseName, + () => Book.findOne(filter).lean(), + attributes + ); + + for (let index = 0; index < warmup; index += 1) { + const result = await runOperation(); + if (!result) throw new Error('Warm-up query did not return the seeded document'); + } + + const latenciesMs = []; + let failures = 0; + let successes = 0; + const measurementStartUs = Date.now() * 1000; + const measurementStart = process.hrtime.bigint(); + await runBounded(operations, concurrency, async () => { + const operationStart = process.hrtime.bigint(); + try { + const result = await runOperation(); + if (!result) throw new Error('Measured query did not return the seeded document'); + successes += 1; + } catch (error) { + failures += 1; + } finally { + latenciesMs.push(Number(process.hrtime.bigint() - operationStart) / 1e6); + } + }); + const elapsedSeconds = Number(process.hrtime.bigint() - measurementStart) / 1e9; + const measurementEndUs = Date.now() * 1000; + latenciesMs.sort((left, right) => left - right); + + writeJsonAtomic(resultFile, { + schemaVersion: 1, + adapter, + benchmarkId, + phase, + workload, + tracingEnabled, + requestedQueryBytes, + actualQueryBytes: actualBytes, + warmupOperations: warmup, + measuredOperations: operations, + successfulOperations: successes, + failedOperations: failures, + concurrency, + measurementStartUs, + measurementEndUs, + elapsedSeconds, + throughputOperationsPerSecond: operations / elapsedSeconds, + latencyMs: { + minimum: latenciesMs[0], + mean: latenciesMs.reduce((sum, value) => sum + value, 0) / latenciesMs.length, + p50: percentile(latenciesMs, 50), + p95: percentile(latenciesMs, 95), + p99: percentile(latenciesMs, 99), + maximum: latenciesMs.at(-1), + }, + runtime: { + node: process.version, + mongoose: mongoose.version, + platform: `${os.platform()}-${os.arch()}`, + }, + telemetry: { + serviceName: process.env.OTEL_SERVICE_NAME || null, + operationName: 'mongoose.benchmarkFind', + }, + database: { + name: databaseName, + directConnection: true, + tls: true, + }, + }); + + await Book.deleteMany({ author: seedAuthor }); + if (failures > 0) process.exitCode = 1; +} + +async function shutdown() { + await mongoose.connection.close(false).catch(() => {}); + await shutdownTelemetry().catch(() => {}); +} + +let shuttingDown = false; +for (const signal of ['SIGINT', 'SIGTERM']) { + process.on(signal, async () => { + if (shuttingDown) return; + shuttingDown = true; + await shutdown(); + process.exit(128 + (signal === 'SIGINT' ? 2 : 15)); + }); +} + +main() + .catch((error) => { + console.error(`Mongoose benchmark failed: ${error.message}`); + process.exitCode = 1; + }) + .finally(shutdown); \ No newline at end of file diff --git a/playgrounds/mongoose/app/package.json b/playgrounds/mongoose/app/package.json index c62ee8d..8c7afcc 100644 --- a/playgrounds/mongoose/app/package.json +++ b/playgrounds/mongoose/app/package.json @@ -8,6 +8,7 @@ "node": "^18.19.0 || >=20.6.0" }, "scripts": { + "benchmark": "node benchmark.js", "start": "node server.js", "test:crud": "node mongoose-crud-test.js" }, diff --git a/playgrounds/mongoose/app/telemetry.js b/playgrounds/mongoose/app/telemetry.js index 1851944..cbed407 100644 --- a/playgrounds/mongoose/app/telemetry.js +++ b/playgrounds/mongoose/app/telemetry.js @@ -90,7 +90,7 @@ function installMongooseTracePropagation(mongoose) { propagationInstalled = true; } -async function traceDocumentDbOperation(operationName, namespace, operation) { +async function traceDocumentDbOperation(operationName, namespace, operation, attributes = {}) { if (!tracingEnabled) { return operation(); } @@ -103,6 +103,7 @@ async function traceDocumentDbOperation(operationName, namespace, operation) { 'db.system.name': 'documentdb', 'db.operation.name': operationName, 'db.namespace': namespace, + ...attributes, }, }, async (span) => { diff --git a/playgrounds/mongoose/scripts/run-telemetry-demo.sh b/playgrounds/mongoose/scripts/run-telemetry-demo.sh index 481b158..5289191 100755 --- a/playgrounds/mongoose/scripts/run-telemetry-demo.sh +++ b/playgrounds/mongoose/scripts/run-telemetry-demo.sh @@ -14,12 +14,12 @@ if [ -f "$ENV_FILE" ]; then set +a fi +export DOCUMENTDB_IMAGE="${DOCUMENTDB_IMAGE:-ghcr.io/documentdb/documentdb/documentdb-local:trace-4fbbfcb8}" "$TELEMETRY_DIR/scripts/up.sh" export OTEL_TRACES_ENABLED=true export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:${OTEL_COLLECTOR_GRPC_PORT:-4317}" export OTEL_SERVICE_NAME="${OTEL_SERVICE_NAME:-documentdb-mongoose}" -export DOCUMENTDB_IMAGE="${DOCUMENTDB_IMAGE:-ghcr.io/documentdb/documentdb/documentdb-local:trace-4fbbfcb8}" echo "" echo "Application tracing is enabled." diff --git a/shared/telemetry/README.md b/shared/telemetry/README.md index cd2b0cc..2ab43af 100644 --- a/shared/telemetry/README.md +++ b/shared/telemetry/README.md @@ -93,3 +93,19 @@ are joined successfully. Other playgrounds can use the same stack by starting it here first and then running their normal application or test scripts. + +## Run a benchmark + +The [local latency benchmark](../../playgrounds/benchmarks/local-latency/) +reuses this stack with `otel-collector-benchmark.yaml`, which omits the verbose +debug exporter during measured runs. It owns experiment artifacts and invokes a +driver adapter: + +```bash +cd ../.. +BENCHMARK_ADAPTER=mongoose ./playgrounds/benchmarks/local-latency/benchmark.sh +``` + +The infrastructure is reusable by every driver playground. Connected tracing +also requires a driver adapter that creates client spans and injects W3C trace +context into each MongoDB command comment. diff --git a/shared/telemetry/compose.yaml b/shared/telemetry/compose.yaml index bed3841..cae58d3 100644 --- a/shared/telemetry/compose.yaml +++ b/shared/telemetry/compose.yaml @@ -16,7 +16,7 @@ services: - "127.0.0.1:${OTEL_COLLECTOR_HTTP_PORT:-4318}:4318" - "127.0.0.1:${OTEL_COLLECTOR_HEALTH_PORT:-13133}:13133" volumes: - - ./otel-collector.yaml:/etc/otelcol-contrib/config.yaml:ro + - ${OTEL_COLLECTOR_CONFIG_FILE:-./otel-collector.yaml}:/etc/otelcol-contrib/config.yaml:ro - documentdb-logs:/var/log/documentdb:ro documentdb: diff --git a/shared/telemetry/otel-collector-benchmark.yaml b/shared/telemetry/otel-collector-benchmark.yaml new file mode 100644 index 0000000..90c0225 --- /dev/null +++ b/shared/telemetry/otel-collector-benchmark.yaml @@ -0,0 +1,35 @@ +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + http: + endpoint: 0.0.0.0:4318 + +processors: + batch: + timeout: 1s + send_batch_size: 1024 + resource: + attributes: + - key: deployment.environment + value: local-benchmark + action: upsert + +exporters: + otlp_grpc/jaeger: + endpoint: jaeger:4317 + tls: + insecure: true + +extensions: + health_check: + endpoint: 0.0.0.0:13133 + +service: + extensions: [health_check] + pipelines: + traces: + receivers: [otlp] + processors: [resource, batch] + exporters: [otlp_grpc/jaeger] \ No newline at end of file From 2da226b468abf7b575313fc518a2750256932603 Mon Sep 17 00:00:00 2001 From: Rayhan Hossain Date: Wed, 12 Aug 2026 15:42:08 -0700 Subject: [PATCH 2/2] address feedback for telemetry env and postgres.execute Signed-off-by: Rayhan Hossain --- .../local-latency/analyze_jaeger.py | 29 +++++++++++++++++++ .../benchmarks/local-latency/benchmark.sh | 11 +++++++ .../local-latency/test_analyze_jaeger.py | 26 +++++++++++++++++ 3 files changed, 66 insertions(+) diff --git a/playgrounds/benchmarks/local-latency/analyze_jaeger.py b/playgrounds/benchmarks/local-latency/analyze_jaeger.py index 2ee571a..cff065b 100755 --- a/playgrounds/benchmarks/local-latency/analyze_jaeger.py +++ b/playgrounds/benchmarks/local-latency/analyze_jaeger.py @@ -62,6 +62,31 @@ def child_of(span, parent_span_id): ) +def is_descendant(span, ancestor_span_id, spans_by_id, max_depth=64): + # Walk CHILD_OF links up to the selected gateway so sibling branches + # (retries, multi-command traces) are not mixed into the metrics. + current = span + seen = set() + for _ in range(max_depth): + parent_id = next( + ( + reference.get("spanID") + for reference in current.get("references", []) + if reference.get("refType") == "CHILD_OF" + ), + None, + ) + if parent_id is None or parent_id in seen: + return False + if parent_id == ancestor_span_id: + return True + seen.add(parent_id) + current = spans_by_id.get(parent_id) + if current is None: + return False + return False + + def complete_trace_count(traces, result, service_name, gateway_service): count = 0 for trace in traces: @@ -98,9 +123,11 @@ def service(span): ) if gateway is None: continue + spans_by_id = {span.get("spanID"): span for span in spans} if any( service(span) == gateway_service and span.get("operationName") == "postgres.execute" + and is_descendant(span, gateway.get("spanID"), spans_by_id) for span in spans ): count += 1 @@ -298,11 +325,13 @@ def service(span): if not child_of(gateway, application.get("spanID")): excluded["unexpected_parent_relationship"] += 1 continue + spans_by_id = {span.get("spanID"): span for span in spans} postgres = [ span for span in spans if service(span) == args.gateway_service and span.get("operationName") == "postgres.execute" + and is_descendant(span, gateway.get("spanID"), spans_by_id) ] if not postgres: excluded["missing_postgres_span"] += 1 diff --git a/playgrounds/benchmarks/local-latency/benchmark.sh b/playgrounds/benchmarks/local-latency/benchmark.sh index 01a8195..e64ec0e 100755 --- a/playgrounds/benchmarks/local-latency/benchmark.sh +++ b/playgrounds/benchmarks/local-latency/benchmark.sh @@ -5,6 +5,17 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" REPO_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" TELEMETRY_DIR="$REPO_ROOT/shared/telemetry" + +# Load shared telemetry settings so custom ports/credentials reach the endpoint +# derivation, the adapter, and the analyzer; up.sh only sees them in its own process. +ENV_FILE="${TELEMETRY_ENV_FILE:-$TELEMETRY_DIR/.env}" +if [ -f "$ENV_FILE" ]; then + set -a + # shellcheck disable=SC1090 + source "$ENV_FILE" + set +a +fi + ADAPTER="${BENCHMARK_ADAPTER:-mongoose}" ADAPTER_SCRIPT="$SCRIPT_DIR/adapters/$ADAPTER.sh" BENCHMARK_ID="${BENCHMARK_ID:-$(date -u +%Y%m%dT%H%M%SZ)-$ADAPTER-$$}" diff --git a/playgrounds/benchmarks/local-latency/test_analyze_jaeger.py b/playgrounds/benchmarks/local-latency/test_analyze_jaeger.py index 97c4f83..1f95aa4 100644 --- a/playgrounds/benchmarks/local-latency/test_analyze_jaeger.py +++ b/playgrounds/benchmarks/local-latency/test_analyze_jaeger.py @@ -4,6 +4,7 @@ build_markdown_summary, complete_trace_count, interval_union_duration, + is_descendant, percentile, ) @@ -82,6 +83,7 @@ def test_complete_trace_count_requires_connected_postgres_span(self): }, { "processID": "gateway", + "spanID": "gateway-span", "operationName": "gateway.request", "references": [ {"refType": "CHILD_OF", "spanID": "client-span"} @@ -90,6 +92,9 @@ def test_complete_trace_count_requires_connected_postgres_span(self): { "processID": "gateway", "operationName": "postgres.execute", + "references": [ + {"refType": "CHILD_OF", "spanID": "gateway-span"} + ], }, ], } @@ -108,6 +113,27 @@ def test_complete_trace_count_requires_connected_postgres_span(self): 0, ) + def test_is_descendant_restricts_to_selected_gateway_branch(self): + spans = [ + {"spanID": "app"}, + {"spanID": "gwA", "references": [{"refType": "CHILD_OF", "spanID": "app"}]}, + {"spanID": "pgA", "references": [{"refType": "CHILD_OF", "spanID": "gwA"}]}, + {"spanID": "gwB", "references": [{"refType": "CHILD_OF", "spanID": "app"}]}, + {"spanID": "pgB", "references": [{"refType": "CHILD_OF", "spanID": "gwB"}]}, + ] + spans_by_id = {span["spanID"]: span for span in spans} + + self.assertTrue(is_descendant(spans_by_id["pgA"], "gwA", spans_by_id)) + self.assertFalse(is_descendant(spans_by_id["pgB"], "gwA", spans_by_id)) + + def test_is_descendant_survives_reference_cycles(self): + cyclic = [ + {"spanID": "x", "references": [{"refType": "CHILD_OF", "spanID": "y"}]}, + {"spanID": "y", "references": [{"refType": "CHILD_OF", "spanID": "x"}]}, + ] + cyclic_by_id = {span["spanID"]: span for span in cyclic} + self.assertFalse(is_descendant(cyclic_by_id["x"], "gwA", cyclic_by_id)) + if __name__ == "__main__": unittest.main() \ No newline at end of file