diff --git a/.github/workflows/collector-experiment.yml b/.github/workflows/collector-experiment.yml new file mode 100644 index 0000000..9bbb7cb --- /dev/null +++ b/.github/workflows/collector-experiment.yml @@ -0,0 +1,46 @@ +name: Collector fault experiment +on: + workflow_dispatch: + pull_request: + paths: + - 'experiments/collector/**' + - '.github/workflows/collector-experiment.yml' +permissions: + contents: read +jobs: + collector: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 + with: + python-version: '3.12' + - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 # v7.1.0 + with: + node-version: '22' + - name: Install experiment dependencies and verify collector + run: | + python -m pip install -r experiments/collector/environment.txt + (cd packages/typescript && npm ci --ignore-scripts && npm run build) + (cd experiments/collector && npm ci --ignore-scripts) + curl -fsSL https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.163.2/otelcol_0.163.2_linux_amd64.tar.gz -o "$RUNNER_TEMP/otelcol.tar.gz" + echo "9fbf3101331c88aab4b7f4cfd7c788b252e91e193793e39bd2154971c87f88a7 $RUNNER_TEMP/otelcol.tar.gz" | sha256sum --check + tar -xzf "$RUNNER_TEMP/otelcol.tar.gz" -C "$RUNNER_TEMP" otelcol + - name: Run synthetic loopback delivery cases + env: + PYTHONPATH: src + run: python experiments/collector/run.py --collector "$RUNNER_TEMP/otelcol" --output collector-results + - name: Run observer controls and stress cases + env: + PYTHONPATH: src + run: | + python -m unittest discover -s experiments/collector -p 'test_*.py' + python experiments/collector/stress.py --collector "$RUNNER_TEMP/otelcol" --output collector-stress-results + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.0 + if: always() + with: + name: collector-experiment-evidence + path: | + collector-results + collector-stress-results diff --git a/experiments/collector/README.md b/experiments/collector/README.md new file mode 100644 index 0000000..36ff964 --- /dev/null +++ b/experiments/collector/README.md @@ -0,0 +1,150 @@ +# Collector fault harness + +This experimental harness wires the real Python and TypeScript telemetry SDKs +through a real OpenTelemetry Collector to a synthetic loopback receiver backed by +SQLite. It is not production delivery evidence and is not a release gate yet. + +## Reproduction + +Use Linux amd64, Python 3.12 and Node >=22.15. From the repository root: + +```sh +uv venv --python 3.12 ../collector-env +uv pip install --python ../collector-env/bin/python -r experiments/collector/environment.txt +(cd packages/typescript && npm ci --ignore-scripts && npm run build) +(cd experiments/collector && npm ci --ignore-scripts) +PYTHONPATH=src ../collector-env/bin/python experiments/collector/run.py --collector /absolute/path/to/otelcol --output ../collector-results-new +``` + +Use the official core collector v0.163.2 Linux amd64 release: +https://github.com/open-telemetry/opentelemetry-collector-releases/releases/tag/v0.163.2 + +Archive `otelcol_0.163.2_linux_amd64.tar.gz` SHA256: +`9fbf3101331c88aab4b7f4cfd7c788b252e91e193793e39bd2154971c87f88a7`. +The runner enforces binary SHA256 +`d4fa834fae73cb37e6058955a3833894d1ec815a1fc3529b0c2aaa99d8b22a2e`. +Use a fresh output directory. Clear HTTP(S)/ALL proxy environment variables for +the Python emitter if your environment exports them; child processes clear these. + +## Assertions validated on an Ubuntu CI host + +For both SDKs: healthy delivery, first-response 503, commit then lost ACK, partial +rejection, duplicate event ID, and collector idle restart. SQLite stores local +evidence, observed requests, and committed rows separately. The synthetic backend +uses a unique constraint for idempotency; the test must not be interpreted as a +collector exactly-once guarantee. Retry observations cover the whole pipeline, +not a unique attribution to collector retry versus exporter retry. The restart +case restarts an idle collector before emission; it does not test in-flight recovery. +In these twelve baseline cases queues are disabled and assertions compare row counts. +The eight stress cases below add queue pressure, reordering and full projection checks. + +## Validation and local host limitation + +Python compilation, JavaScript syntax, dependency installation, and the repository +TypeScript build succeeded. The downloaded archive matched the release digest and +`otelcol components` succeeded. The host then failed before receiver startup: + +``` +Error: failed to register process metrics: process does not exist +``` + +Setting `service.telemetry.metrics.level: none` did not bypass the error. Upstream +collector v0.163.0 `service/service.go` calls `registerProcessMetrics` on Linux +regardless of that setting; `service/internal/proctelemetry/process_telemetry.go` +initializes gopsutil with `os.Getpid()`. No collector delivery case passed on the local sandbox host. The supported Ubuntu +CI run below completed all twelve cases after correcting the SDK v3 processor +constructor. No production endpoint or separately provisioned cloud resource was used. + +## Recorded supported-host run + +[Run 39665288508](https://github.com/agentrust-io/agentrust-telemetry/actions/runs/39665288508), +head `42fa91f0194741a1a1c7102d08142e7925fc86cb`, Ubuntu 24.04, +Python 3.12.15 and Node 22.23.3: all twelve cases passed. For both languages: + +| Case | Local evidence rows | Observed attempts | Backend commits | +| --- | ---: | ---: | ---: | +| Healthy | 1 | 1 | 1 | +| First 503 | 1 | 2 | 1 | +| Lost acknowledgement | 1 | 2 | 1 | +| Partial rejection | 1 | 1 | 0 | +| Duplicate event | 2 | 2 | 1 | +| Idle restart | 1 | 1 | 1 | + +The run retains configs, collector/emitter logs, synthetic inputs, SQLite observer +and per-case summary in the `collector-experiment-evidence` artifact. These +measurements establish the listed synthetic pipeline behaviors only. + +## Added stress and fidelity checks (new head) + +`stress.py` adds four cases for each SDK: an eight-event healthy queue control, +an eight-event blocked-consumer/one-slot queue saturation case, a collector kill +between backend commit and acknowledgement followed by a fresh post-restart event, +and three events emitted with descending source timestamps. The sending queue is +in memory; this does not claim persistent queue recovery. The receiver retains +complete normalized bodies, event names and nanosecond timestamps in SQLite. +Every observed body is compared to its submitted event, and every non-saturation +case must account for all committed event IDs. Saturation requires both positive +commits and independently observed loss relative to local evidence. + +The TypeScript experiment emitter now passes the normalized event timestamp to +the OTel SDK as an integer seconds/nanoseconds pair. This is experiment wiring, +not a change to the reference SDK or a claim about all exporters/backends. + +Six local observer tests catch changed nested fields, numeric type changes, +wrong event names/timestamps, missing observations, dropped local evidence, +a non-descending input control and a vacuous saturation result. The pinned collector validates +the generated configuration. Local collector startup still fails at process +metrics registration before any delivery case. The supported-host run below +validates the added cases; the twelve-case historical run above remains evidence +only for its original head. + +## Recorded stress run + +[Run 40204045774](https://github.com/agentrust-io/agentrust-telemetry/actions/runs/40204045774), +code head `02d5c64572ee8341f3a4d54c10e8a400a51d76f8`, Ubuntu, +Python 3.12.15: all twelve baseline cases, six observer unit tests, and all eight +stress cases passed. Each language produced the same counts: + +| Case | Local evidence | Observed attempts | Backend commits | +| --- | ---: | ---: | ---: | +| Healthy queue | 8 | 8 | 8 | +| Saturation | 8 | 1 | 1 | +| In-flight restart | 2 | 2 | 2 | +| Reversed timestamp input | 3 | 3 | 3 | + +Every observed stress record retained its complete normalized body, event name +and nanosecond timestamp. Saturation waits for the seeded request to reach the +blocked receiver before submitting pressure. Collector 0.163.2 counts that +in-flight request against this one-slot queue capacity, so the seeded event alone +committed; the runner also requires the Collector's explicit queue-full log. +These are observed counts for this run, not universal retry/delivery guarantees. + +The first two stress workflow attempts exposed harness assumptions about consumer +scheduling and queue capacity. Both were corrected before the successful run. +The artifact includes full inputs, configs, Collector/emitter logs, SQLite +observations and both summary files. No runtime delivery policy was changed. + +## Transport-order control correction + +The documentation confirmation run 40204226853 observed all three TypeScript +reversed-input events with intact bodies/timestamps, but their concurrent HTTP +exports arrived in reverse order of emission. The original arrival-order +assertion was too strong and failed. The corrected harness verifies descending +input timestamps, the exact input event-ID sequence in local evidence, complete +backend delivery and per-event projection fidelity. It records submitted and +observed event-ID sequences separately, allowing transport reordering and +retries. It does not claim an SDK or Collector arrival-order guarantee. The +recorded twenty-case run above predates this observer correction. + +[Run 40204492721](https://github.com/agentrust-io/agentrust-telemetry/actions/runs/40204492721) +confirmed the corrected observer contract on code head +`3cf135ad1698d9157e99ff1396e8c1665bb0b4a3`: all twelve baseline cases, +six observer unit tests and all eight stress cases passed on Ubuntu. This +records evidence for that code head; this documentation update adds no new +execution result. In the baseline `lost_ack` case, the receiver commits before +dropping the acknowledgement, then a retry of the same event identity yields +one backend commit across multiple attempts. That is synthetic backend +deduplication, not an exactly-once Collector guarantee. The stress +`inflight_restart` case commits before acknowledgement, kills the Collector, +and emits a different event after restart; it does not replay the unacknowledged +identity or establish persistent queue recovery. diff --git a/experiments/collector/emit.mjs b/experiments/collector/emit.mjs new file mode 100644 index 0000000..4189e67 --- /dev/null +++ b/experiments/collector/emit.mjs @@ -0,0 +1,24 @@ +import fs from 'node:fs'; +import {LoggerProvider, SimpleLogRecordProcessor} from '@opentelemetry/sdk-logs'; +import {OTLPLogExporter} from '@opentelemetry/exporter-logs-otlp-http'; +import {SchemaValidator, TelemetryClient} from '../../packages/typescript/dist/index.js'; +const [endpoint, input, evidence, timeout = '4000'] = process.argv.slice(2); +const provider = new LoggerProvider({processors: [new SimpleLogRecordProcessor({exporter: new OTLPLogExporter({url: endpoint, timeoutMillis: Number(timeout)})})]}); +const logger = provider.getLogger('agentrust-collector-acceptance'); +const client = new TelemetryClient(SchemaValidator.bundled(), { + evidenceSink: {append(event) { + const fd = fs.openSync(evidence, 'a'); + try {fs.writeSync(fd, JSON.stringify(event) + '\n'); fs.fsyncSync(fd);} finally {fs.closeSync(fd);} + }}, + logEmitter: {emit(record) { + const ns = BigInt(record.timestamp_ns); + logger.emit({body: record.body, eventName: record.event_name, + timestamp: [Number(ns / 1_000_000_000n), Number(ns % 1_000_000_000n)]}); + }}, +}); +for (const event of JSON.parse(fs.readFileSync(input, 'utf8'))) { + const result = client.emit(event); + if (!result.accepted || !result.evidencePersisted || !result.logEmitted) throw new Error('Local acceptance failed'); + await provider.forceFlush(); +} +await provider.shutdown(); diff --git a/experiments/collector/environment.txt b/experiments/collector/environment.txt new file mode 100644 index 0000000..87e0354 --- /dev/null +++ b/experiments/collector/environment.txt @@ -0,0 +1,46 @@ +agentrust-trace==0.10.0 +annotated-types==0.8.0 +arrow==1.4.0 +attrs==26.1.0 +build==1.6.1 +certifi==2026.10.19 +cffi==2.1.1 +charset-normalizer==3.5.2 +cryptography==50.0.1 +fqdn==1.5.1 +googleapis-common-protos==1.75.5 +idna==3.19 +isoduration==20.11.0 +jsonpointer==3.1.1 +jsonschema==4.26.0 +jsonschema-specifications==2025.9.1 +opentelemetry-api==1.45.1 +opentelemetry-exporter-http-transport==0.66b1 +opentelemetry-exporter-otlp-common==0.66b1 +opentelemetry-exporter-otlp-proto-common==1.45.1 +opentelemetry-exporter-otlp-proto-http==1.45.1 +opentelemetry-proto==1.45.1 +opentelemetry-sdk==1.45.1 +opentelemetry-semantic-conventions==0.66b1 +packaging==26.3 +protobuf==7.36.2 +pycparser==3.0 +pydantic==2.13.5 +pydantic-core==2.46.5 +pyproject-hooks==1.2.0 +python-dateutil==2.9.0.post0 +pyyaml==6.0.3 +referencing==0.37.0 +requests==2.34.2 +rfc3339-validator==0.1.4 +rfc3986-validator==0.1.1 +rfc3987==1.3.8 +rfc8785==0.1.4 +rpds-py==2026.6.3 +six==1.17.0 +typing-extensions==4.16.0 +typing-inspection==0.4.4 +tzdata==2026.3 +uri-template==1.3.0 +urllib3==2.8.0 +webcolors==25.10.0 diff --git a/experiments/collector/fidelity.py b/experiments/collector/fidelity.py new file mode 100644 index 0000000..d72d0e1 --- /dev/null +++ b/experiments/collector/fidelity.py @@ -0,0 +1,49 @@ +"""Pure observer assertions; local acceptance, attempts and commits stay distinct.""" +import json + + +def any_value(value): + kind = value.WhichOneof("value") + if kind == "kvlist_value": + return {item.key: any_value(item.value) for item in value.kvlist_value.values} + if kind == "array_value": + return [any_value(item) for item in value.array_value.values] + if kind in {"string_value", "bool_value", "int_value", "double_value"}: + return getattr(value, kind) + raise ValueError(f"unsupported normalized-event value {kind}") + + +def check_projection(events, observations): + expected = {event["event_id"]: event for event in events} + if not observations: + raise AssertionError("no independent projection observations") + for observation in observations: + event = observation["body"] + original = expected[event["event_id"]] + if json.dumps(event, sort_keys=True, allow_nan=False) != json.dumps(original, sort_keys=True, allow_nan=False): + raise AssertionError("normalized event body changed") + if observation["event_name"] != original["event_type"]: + raise AssertionError("event name changed") + if observation["timestamp_ns"] != original["time_unix_nano"]: + raise AssertionError("event timestamp changed") + + +def check_delivery(mode, events, evidence_count, attempts, commits): + ids = [event["event_id"] for event in events] + if evidence_count != len(events): + raise AssertionError("local evidence is missing") + if not set(commits).issubset(ids) or not set(attempts).issubset(ids): + raise AssertionError("observer saw an unsubmitted event") + if mode == "queue_saturation": + if not 0 < len(set(commits)) < len(events): + raise AssertionError("expected bounded queue loss was not independently observed") + elif set(commits) != set(ids): + raise AssertionError("a submitted event did not commit") + if mode == "reorder": + times = [int(event["time_unix_nano"]) for event in events] + if len(times) < 2 or not all(left > right for left, right in zip(times, times[1:])): + raise AssertionError("reorder control did not submit descending timestamps") + if set(attempts) != set(ids): + raise AssertionError("a reordered input was not independently observed") + # Concurrent SDK exports/HTTP requests need not preserve arrival order. + # check_projection binds every observed timestamp and body to its event ID. diff --git a/experiments/collector/package-lock.json b/experiments/collector/package-lock.json new file mode 100644 index 0000000..83f52c9 --- /dev/null +++ b/experiments/collector/package-lock.json @@ -0,0 +1,166 @@ +{ + "name": "agentrust-collector-acceptance", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "agentrust-collector-acceptance", + "dependencies": { + "@opentelemetry/exporter-logs-otlp-http": "0.300.0", + "@opentelemetry/sdk-logs": "3.0.0" + } + }, + "node_modules/@opentelemetry/api": { + "version": "1.10.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.10.0.tgz", + "integrity": "sha512-lMkyes8iHrG+mk6WH7X70MdCLI2RS0FwNBY1PyQ8UemaijE2m5mv5A0Niy3HtbYTFrgkUEQjFMXmSAk2hQDuPQ==", + "license": "Apache-2.0", + "peer": true, + "engines": { + "node": ">=8.0.0" + } + }, + "node_modules/@opentelemetry/core": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-3.0.0.tgz", + "integrity": "sha512-whOLlww0R/RLtCWGBXhK1fdvL9t2/TL0fqdjZvcaoUUuNilRPjXecqQdzyVwV68s6XEDA5oYbm7/bZT62GXF4A==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/exporter-logs-otlp-http": { + "version": "0.300.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/exporter-logs-otlp-http/-/exporter-logs-otlp-http-0.300.0.tgz", + "integrity": "sha512-S8yP9Fg33CU9+0fr1pnh/afPfgEpSyxQO4JRUZvmDMhAzgR/RLzvYg93MjWjbC1c2VrXffXROMO68RZ1P/ebgQ==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/otlp-exporter-base": "0.300.0", + "@opentelemetry/otlp-transformer": "0.300.0", + "@opentelemetry/sdk-logs": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.3.0" + } + }, + "node_modules/@opentelemetry/otlp-exporter-base": { + "version": "0.300.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/otlp-exporter-base/-/otlp-exporter-base-0.300.0.tgz", + "integrity": "sha512-D+PhzYURtv6p7Kk1rUrilLtV+ZvS218tI0jiKTV5z1agOeELw78C+cFd3o9UUshUx1j+EIpFM62yGk8BuydQvg==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/otlp-transformer": "0.300.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.3.0" + } + }, + "node_modules/@opentelemetry/otlp-transformer": { + "version": "0.300.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/otlp-transformer/-/otlp-transformer-0.300.0.tgz", + "integrity": "sha512-T55lRmjRwvXWyTFdy/BbJ3T5RRsq3bAwQgxxA7qHFxgRd4+pZzRk02ZwOFsYK5072OXGYjYFNJcp68eQV4i0Ig==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/sdk-logs": "3.0.0", + "@opentelemetry/sdk-metrics": "3.0.0", + "@opentelemetry/sdk-trace": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/resources": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-3.0.0.tgz", + "integrity": "sha512-4762KVmlDroRIt+4Ra8AnwcQqdXlkAR55jDrhzna8bzhJx1mUzvHW/Lh/OemFuY9+ouM4xaDc6CbUmoJmM6BFg==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/sdk-logs": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-logs/-/sdk-logs-3.0.0.tgz", + "integrity": "sha512-IysNZWrBk2PqYw4qU9PZcC7pjrZLa+KLk7vcLmPPNBUr97MgeamWhX0DUHsmxwJv3dnjYM6mLRmP8mw69ld+dA==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/sdk-metrics": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-metrics/-/sdk-metrics-3.0.0.tgz", + "integrity": "sha512-YozyS2rIlizvUNLZwcWxH0x2FhX02/mcZwzM8hsl8yvQk+xeKmAc7vcSieWiO4oSwIB20B8eS/zwdopAd/J5zQ==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/sdk-trace": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace/-/sdk-trace-3.0.0.tgz", + "integrity": "sha512-4T0DbDGdtsjNmYK5/lNalHro4fsMZ0UZ3JE+C8fcqYP25xaJC22FHg8kdJNJ21srjJY5XoT7WNFCtj88NNTl4Q==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "3.0.0", + "@opentelemetry/resources": "3.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": ">=22.15.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.10.0 <1.11.0" + } + }, + "node_modules/@opentelemetry/semantic-conventions": { + "version": "1.43.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/semantic-conventions/-/semantic-conventions-1.43.0.tgz", + "integrity": "sha512-eSYWTm620tTk45EKSedaUL8MFYI8hW164hIXsgIHyxu3VobUB3fFCu5t0hQby6OoWRPsG1KkKUG2M5UadiLiVg==", + "license": "Apache-2.0", + "engines": { + "node": ">=14" + } + } + } +} diff --git a/experiments/collector/package.json b/experiments/collector/package.json new file mode 100644 index 0000000..1c66e5d --- /dev/null +++ b/experiments/collector/package.json @@ -0,0 +1 @@ +{"name":"agentrust-collector-acceptance","private":true,"type":"module","dependencies":{"@opentelemetry/exporter-logs-otlp-http":"0.300.0","@opentelemetry/sdk-logs":"3.0.0"}} \ No newline at end of file diff --git a/experiments/collector/run.py b/experiments/collector/run.py new file mode 100644 index 0000000..c40aa11 --- /dev/null +++ b/experiments/collector/run.py @@ -0,0 +1,195 @@ +"""Real OTLP collector experiment, synthetic events and a local SQLite observer.""" +from __future__ import annotations + +import argparse +import gzip +import hashlib +import json +import os +from pathlib import Path +import socket +import sqlite3 +import subprocess +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from uuid import uuid4 + +from google.protobuf import json_format +from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ( + ExportLogsServiceRequest, ExportLogsServiceResponse, +) +from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import SimpleLogRecordProcessor +from agentrust_telemetry import OTelLogEmitter, SchemaValidator, TelemetryClient + +ROOT = Path(__file__).resolve().parents[2] +BINARY_SHA = 'd4fa834fae73cb37e6058955a3833894d1ec815a1fc3529b0c2aaa99d8b22a2e' + + +def port(): + with socket.socket() as s: + s.bind(('127.0.0.1', 0)) + return s.getsockname()[1] + + +def wait_ready(number, process): + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError('collector exited; inspect collector.log') + try: + with socket.create_connection(('127.0.0.1', number), timeout=.1): + return + except OSError: + time.sleep(.02) + raise TimeoutError('collector readiness') + + +def run(binary, out): + if hashlib.sha256(binary.read_bytes()).hexdigest() != BINARY_SHA: + raise ValueError('collector binary digest differs from pinned 0.163.2 Linux amd64') + out.mkdir(parents=True, exist_ok=False) + database = out / 'observations.sqlite' + with sqlite3.connect(database) as db: + db.executescript('CREATE TABLE attempts(case_id TEXT,event_id TEXT);' + 'CREATE TABLE committed(case_id TEXT,event_id TEXT, PRIMARY KEY(case_id,event_id));' + 'CREATE TABLE evidence(case_id TEXT,event_id TEXT,body TEXT);') + state = {'case': '', 'mode': '', 'requests': 0} + lock = threading.Lock() + + class Receiver(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + body = self.rfile.read(int(self.headers['Content-Length'])) + if self.headers.get('Content-Encoding') == 'gzip': + body = gzip.decompress(body) + request = ExportLogsServiceRequest() + if 'json' in self.headers.get('Content-Type', ''): + json_format.Parse(body.decode(), request) + else: + request.ParseFromString(body) + ids = [next(v.value.string_value for v in record.body.kvlist_value.values + if v.key == 'event_id') + for resource in request.resource_logs for scope in resource.scope_logs + for record in scope.log_records] + with lock: + case, mode = state['case'], state['mode'] + state['requests'] += 1 + number = state['requests'] + with sqlite3.connect(database) as db: + db.executemany('INSERT INTO attempts VALUES (?,?)', [(case, i) for i in ids]) + if not (mode == 'partial' or (mode == 'retry' and number == 1)): + db.executemany('INSERT OR IGNORE INTO committed VALUES (?,?)', [(case, i) for i in ids]) + if mode == 'lost_ack' and number == 1: + self.connection.shutdown(socket.SHUT_RDWR) + self.connection.close() + return + response = ExportLogsServiceResponse() + if mode == 'partial': + response.partial_success.rejected_log_records = len(ids) + response.partial_success.error_message = 'synthetic controlled refusal' + payload = response.SerializeToString() + self.send_response(503 if mode == 'retry' and number == 1 else 200) + self.send_header('Content-Type', 'application/x-protobuf') + self.send_header('Content-Length', str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + receiver = ThreadingHTTPServer(('127.0.0.1', 0), Receiver) + thread = threading.Thread(target=receiver.serve_forever, daemon=True) + thread.start() + summaries = [] + clean_env = {k: v for k, v in os.environ.items() + if k.lower() not in {'http_proxy', 'https_proxy', 'all_proxy'}} + try: + for language in ('python', 'typescript'): + for mode in ('healthy', 'retry', 'lost_ack', 'partial', 'duplicate', 'restart'): + case = f'{language}-{mode}' + directory = out / case + directory.mkdir() + with lock: + state.update(case=case, mode=mode, requests=0) + number = port() + config = directory / 'collector.json' + config.write_text(json.dumps({ + 'receivers': {'otlp': {'protocols': {'http': {'endpoint': f'127.0.0.1:{number}'}}}}, + 'exporters': {'otlp_http': { + 'endpoint': f'http://127.0.0.1:{receiver.server_port}', + 'sending_queue': {'enabled': False}, + 'retry_on_failure': {'enabled': True, 'initial_interval': '50ms', + 'max_interval': '100ms', 'max_elapsed_time': '1s'}, + }}, + 'service': {'telemetry': {'metrics': {'level': 'none'}}, 'pipelines': {'logs': {'receivers': ['otlp'], 'exporters': ['otlp_http']}}}, + }, indent=2)) + event = json.loads((ROOT / 'conformance/fixtures/valid/usage.json').read_text()) + event['event_id'] = str(uuid4()) + events = [event, event] if mode == 'duplicate' else [event] + (directory / 'input.json').write_text(json.dumps(events)) + with (directory / 'collector.log').open('wb') as log: + process = subprocess.Popen([str(binary), '--config', str(config)], + stdout=log, stderr=subprocess.STDOUT, env=clean_env) + try: + wait_ready(number, process) + if mode == 'restart': + process.terminate(); process.wait(timeout=5) + process = subprocess.Popen([str(binary), '--config', str(config)], + stdout=log, stderr=subprocess.STDOUT, env=clean_env) + wait_ready(number, process) + endpoint = f'http://127.0.0.1:{number}/v1/logs' + if language == 'python': + class Evidence: + def append(self, item): + with sqlite3.connect(database) as db: + db.execute('INSERT INTO evidence VALUES (?,?,?)', + (case, item['event_id'], json.dumps(item))) + provider = LoggerProvider() + provider.add_log_record_processor(SimpleLogRecordProcessor( + OTLPLogExporter(endpoint=endpoint, timeout=4))) + client = TelemetryClient(SchemaValidator.bundled(), evidence_sink=Evidence(), + log_emitter=OTelLogEmitter(provider.get_logger('acceptance'))) + try: + for item in events: + result = client.emit(item) + assert result.accepted and result.evidence_persisted and result.log_emitted + provider.force_flush() + finally: + provider.shutdown() + else: + with (directory / 'emitter.log').open('wb') as emitter_log: + subprocess.run(['node', str(Path(__file__).with_name('emit.mjs')), endpoint, + str(directory / 'input.json'), str(directory / 'evidence.ndjson')], + check=True, timeout=15, env=clean_env, + stdout=emitter_log, stderr=subprocess.STDOUT) + rows = [json.loads(x) for x in (directory / 'evidence.ndjson').read_text().splitlines()] + with sqlite3.connect(database) as db: + db.executemany('INSERT INTO evidence VALUES (?,?,?)', + [(case, x['event_id'], json.dumps(x)) for x in rows]) + finally: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill(); process.wait() + with sqlite3.connect(database) as db: + counts = {table: db.execute(f'SELECT count(*) FROM {table} WHERE case_id=?', (case,)).fetchone()[0] + for table in ('evidence', 'attempts', 'committed')} + assert counts['evidence'] == len(events), (case, counts) + assert counts['committed'] == (0 if mode == 'partial' else 1), (case, counts) + assert counts['attempts'] >= (2 if mode in {'retry', 'lost_ack', 'duplicate'} else 1), (case, counts) + summaries.append({'case': case, **counts}) + (out / 'summary.json').write_text(json.dumps(summaries, indent=2) + '\n') + finally: + receiver.shutdown(); receiver.server_close(); thread.join(timeout=5) + return summaries + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--collector', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + args = parser.parse_args() + print(json.dumps(run(args.collector.resolve(), args.output.resolve()), indent=2)) diff --git a/experiments/collector/stress.py b/experiments/collector/stress.py new file mode 100644 index 0000000..3263591 --- /dev/null +++ b/experiments/collector/stress.py @@ -0,0 +1,241 @@ +"""Queue pressure, in-flight collector loss and complete event projection checks.""" +from __future__ import annotations + +import argparse +import gzip +import hashlib +import json +import os +from pathlib import Path +import sqlite3 +import subprocess +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from uuid import uuid4 + +from google.protobuf import json_format +from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ExportLogsServiceRequest, ExportLogsServiceResponse +from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import SimpleLogRecordProcessor +from agentrust_telemetry import OTelLogEmitter, SchemaValidator, TelemetryClient + +from fidelity import any_value, check_delivery, check_projection +from run import BINARY_SHA, ROOT, port, wait_ready + + +def wait_until(predicate, description): + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(.01) + raise TimeoutError(description) + + +def run(binary, output): + if hashlib.sha256(binary.read_bytes()).hexdigest() != BINARY_SHA: + raise ValueError("collector binary differs from pinned release") + output.mkdir(parents=True, exist_ok=False) + database = output / "observations.sqlite" + with sqlite3.connect(database) as db: + db.executescript("CREATE TABLE evidence(case_id TEXT,body TEXT);" + "CREATE TABLE attempts(case_id TEXT,event_id TEXT,body TEXT);" + "CREATE TABLE committed(case_id TEXT,event_id TEXT,PRIMARY KEY(case_id,event_id));") + state = {} + lock = threading.Lock() + entered, release = threading.Event(), threading.Event() + + class Receiver(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + raw = self.rfile.read(int(self.headers['Content-Length'])) + if self.headers.get('Content-Encoding') == 'gzip': + raw = gzip.decompress(raw) + request = ExportLogsServiceRequest() + if 'json' in self.headers.get('Content-Type', ''): + json_format.Parse(raw.decode('utf-8'), request) + else: + request.ParseFromString(raw) + observations = [{"body": any_value(record.body), "event_name": record.event_name, + "timestamp_ns": str(record.time_unix_nano)} + for resource in request.resource_logs for scope in resource.scope_logs + for record in scope.log_records] + with lock: + case, mode = state['case'], state['mode'] + state['requests'] += 1 + first = state['requests'] == 1 + with sqlite3.connect(database) as db: + db.executemany('INSERT INTO attempts VALUES (?,?,?)', + [(case, item['body']['event_id'], json.dumps(item)) for item in observations]) + if first and mode == 'queue_saturation': + entered.set() + if not release.wait(30): + raise TimeoutError('test never released blocked backend') + with sqlite3.connect(database) as db: + db.executemany('INSERT OR IGNORE INTO committed VALUES (?,?)', + [(case, item['body']['event_id']) for item in observations]) + if first and mode == 'inflight_restart': + entered.set() + if not release.wait(30): + raise TimeoutError('test never released lost acknowledgement') + payload = ExportLogsServiceResponse().SerializeToString() + try: + self.send_response(200) + self.send_header('Content-Type', 'application/x-protobuf') + self.send_header('Content-Length', str(len(payload))) + self.end_headers() + self.wfile.write(payload) + except (BrokenPipeError, ConnectionResetError): + pass # Expected after the collector is killed before the ACK. + + receiver = ThreadingHTTPServer(('127.0.0.1', 0), Receiver) + receiver.daemon_threads = True + thread = threading.Thread(target=receiver.serve_forever, daemon=True) + thread.start() + clean_env = {key: value for key, value in os.environ.items() + if key.lower() not in {'http_proxy', 'https_proxy', 'all_proxy'}} + summaries = [] + + def read_rows(table, column, case): + with sqlite3.connect(database) as db: + return [row[0] for row in db.execute(f'SELECT {column} FROM {table} WHERE case_id=? ORDER BY rowid', (case,))] + + try: + for language in ('python', 'typescript'): + for mode in ('queue_healthy', 'queue_saturation', 'inflight_restart', 'reorder'): + case = f'{language}-{mode}' + directory = output / case + directory.mkdir() + entered.clear() + release.clear() + with lock: + state.update(case=case, mode=mode, requests=0) + number = port() + queue = {'enabled': mode != 'reorder', 'num_consumers': 1, + 'queue_size': 1, 'block_on_overflow': False} + config = directory / 'collector.json' + config.write_text(json.dumps({ + 'receivers': {'otlp': {'protocols': {'http': {'endpoint': f'127.0.0.1:{number}'}}}}, + 'exporters': {'otlp_http': {'endpoint': f'http://127.0.0.1:{receiver.server_port}', + 'timeout': '30s', 'sending_queue': queue, 'retry_on_failure': {'enabled': False}}}, + 'service': {'telemetry': {'metrics': {'level': 'none'}}, 'pipelines': { + 'logs': {'receivers': ['otlp'], 'exporters': ['otlp_http']}}}}, indent=2)) + count = 8 if mode.startswith('queue_') else (2 if mode == 'inflight_restart' else 3) + events = [] + for index in range(count): + event = json.loads((ROOT / 'conformance/fixtures/valid/usage.json').read_text()) + event['event_id'] = str(uuid4()) + event['time_unix_nano'] = str(int(event['time_unix_nano']) + index) + events.append(event) + if mode == 'reorder': + events.reverse() + (directory / 'input.json').write_text(json.dumps(events)) + endpoint = f'http://127.0.0.1:{number}/v1/logs' + + def emit(items, phase): + if language == 'python': + class Evidence: + def append(self, item): + with sqlite3.connect(database) as db: + db.execute('INSERT INTO evidence VALUES (?,?)', (case, json.dumps(item))) + provider = LoggerProvider() + provider.add_log_record_processor(SimpleLogRecordProcessor(OTLPLogExporter(endpoint=endpoint, timeout=.25))) + client = TelemetryClient(SchemaValidator.bundled(), evidence_sink=Evidence(), + log_emitter=OTelLogEmitter(provider.get_logger('stress'))) + try: + for item in items: + result = client.emit(item) + if not result.accepted or not result.evidence_persisted: + raise AssertionError('local acceptance failed') + provider.force_flush() + finally: + provider.shutdown() + else: + input_path = directory / f'{phase}-input.json' + evidence_path = directory / f'{phase}-evidence.ndjson' + input_path.write_text(json.dumps(items)) + with (directory / f'{phase}-emitter.log').open('wb') as log: + subprocess.run(['node', str(Path(__file__).with_name('emit.mjs')), endpoint, + str(input_path), str(evidence_path), '250'], check=True, timeout=30, + env=clean_env, stdout=log, stderr=subprocess.STDOUT) + with sqlite3.connect(database) as db: + db.executemany('INSERT INTO evidence VALUES (?,?)', + [(case, line) for line in evidence_path.read_text().splitlines()]) + + with (directory / 'collector.log').open('wb') as log: + def start(): + child = subprocess.Popen([str(binary), '--config', str(config)], stdout=log, + stderr=subprocess.STDOUT, env=clean_env) + try: + wait_ready(number, child) + except BaseException: + child.kill() + child.wait(timeout=5) + raise + return child + process = start() + try: + if mode == 'queue_healthy': + for index, event in enumerate(events): + emit([event], str(index)) + wait_until(lambda: event['event_id'] in read_rows('committed', 'event_id', case), 'healthy queue drain') + elif mode == 'inflight_restart': + emit(events[:1], 'before') + if not entered.wait(5): + raise TimeoutError('backend did not reach committed-before-ACK boundary') + process.kill() + process.wait(timeout=5) + release.set() + process = start() + emit(events[1:], 'after') + elif mode == 'queue_saturation': + emit(events[:1], 'seed') + if not entered.wait(5): + raise TimeoutError('saturation never blocked a consumer') + emit(events[1:], 'pressure') + release.set() + else: + emit(events, 'all') + minimum = 1 if mode == 'queue_saturation' else len(events) + wait_until(lambda: len(read_rows('committed', 'event_id', case)) >= minimum, f'{case}: backend completion (see SQLite observations)') + finally: + release.set() + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + if mode == 'queue_saturation' and 'sending queue is full' not in (directory / 'collector.log').read_text(): + raise AssertionError(f'{case}: collector did not report queue overflow') + observations = [json.loads(body) for body in read_rows('attempts', 'body', case)] + evidence = [json.loads(body) for body in read_rows('evidence', 'body', case)] + evidence_count = len(evidence) + if mode == 'reorder' and [item['event_id'] for item in evidence] != [item['event_id'] for item in events]: + raise AssertionError(f'{case}: local evidence did not retain reversed input order') + attempts = read_rows('attempts', 'event_id', case) + commits = read_rows('committed', 'event_id', case) + check_projection(events, observations) + check_delivery(mode, events, evidence_count, attempts, commits) + summaries.append({'case': case, 'evidence': evidence_count, 'attempts': len(attempts), + 'committed': len(commits), 'body_and_timestamp_fidelity': True, + 'submitted_ids': [item['event_id'] for item in events], 'observed_ids': attempts}) + (output / 'summary.json').write_text(json.dumps(summaries, indent=2) + '\n') + finally: + release.set() + receiver.shutdown() + receiver.server_close() + thread.join(timeout=5) + return summaries + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--collector', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + args = parser.parse_args() + print(json.dumps(run(args.collector.resolve(), args.output.resolve()), indent=2)) diff --git a/experiments/collector/test_fidelity.py b/experiments/collector/test_fidelity.py new file mode 100644 index 0000000..dc0c73f --- /dev/null +++ b/experiments/collector/test_fidelity.py @@ -0,0 +1,66 @@ +import copy +import unittest + +from opentelemetry.proto.common.v1.common_pb2 import AnyValue + +from fidelity import any_value, check_delivery, check_projection + + +class ObserverTests(unittest.TestCase): + def setUp(self): + self.event = {'event_id': 'one', 'event_type': 'usage.recorded', + 'time_unix_nano': '1787079590000000001', 'input_tokens': 7, + 'cost': {'amount': .0124, 'currency': 'USD'}} + self.observation = {'body': copy.deepcopy(self.event), 'event_name': 'usage.recorded', + 'timestamp_ns': self.event['time_unix_nano']} + + def test_positive_projection(self): + check_projection([self.event], [self.observation]) + + def test_independent_field_changes_cannot_hide_behind_event_id(self): + for change in ('body', 'name', 'timestamp', 'numeric_type'): + item = copy.deepcopy(self.observation) + if change == 'body': + item['body']['cost']['currency'] = 'EUR' + elif change == 'name': + item['event_name'] = 'other' + elif change == 'timestamp': + item['timestamp_ns'] = str(int(item['timestamp_ns']) + 1) + else: + item['body']['input_tokens'] = 7.0 + with self.subTest(change=change), self.assertRaises(AssertionError): + check_projection([self.event], [item]) + + def test_empty_observations_are_not_fidelity_evidence(self): + with self.assertRaises(AssertionError): + check_projection([self.event], []) + + def test_nested_protobuf_values_preserve_types(self): + value = AnyValue() + item = value.kvlist_value.values.add(key='n') + item.value.int_value = 7 + item = value.kvlist_value.values.add(key='flag') + item.value.bool_value = True + self.assertEqual(any_value(value), {'n': 7, 'flag': True}) + self.assertIs(type(any_value(value)['n']), int) + self.assertIs(type(any_value(value)['flag']), bool) + + def test_reordered_input_preserves_events_without_transport_order_guarantee(self): + events = [self.event, {**self.event, 'event_id': 'two', + 'time_unix_nano': '1787079590000000000'}] + check_delivery('reorder', events, 2, ['one', 'two'], ['one', 'two']) + check_delivery('reorder', events, 2, ['two', 'one'], ['one', 'two']) + with self.assertRaises(AssertionError): + check_delivery('reorder', list(reversed(events)), 2, ['one', 'two'], ['one', 'two']) + for evidence, attempts, commits in ((1, ['one', 'two'], ['one', 'two']), + (2, ['one'], ['one', 'two']), + (2, ['one'], ['one'])): + with self.subTest(evidence=evidence, attempts=attempts, commits=commits), self.assertRaises(AssertionError): + check_delivery('reorder', events, evidence, attempts, commits) + + def test_saturation_requires_observed_loss_with_positive_commits(self): + events = [self.event, {**self.event, 'event_id': 'two'}] + check_delivery('queue_saturation', events, 2, ['one'], ['one']) + for commits in ([], ['one', 'two']): + with self.subTest(commits=commits), self.assertRaises(AssertionError): + check_delivery('queue_saturation', events, 2, commits, commits)