@crvouga/mockingbird-service-otel

Stateful mock of an OTLP/HTTP collector in front of the OpenObserve (O2) search API, over one store. Local and E2E runs stop exporting to production telemetry infrastructure, and a test can assert on structured events: emit through the real OpenTelemetry SDK, then read the log back with the same SQL our ops feed and investigation agent send ("the reconcile cron emitted initial_credit_reconcile_completed with granted=1").

• Operation coverage: SUPPORT.md
• OTLP/HTTP follows opentelemetry-proto v1; the O2 routes follow OpenObserve's API, trimmed to what our clients call.

Install

npm install -D @crvouga/mockingbird-service-otel

ESM only. Node >= 22 or Bun >= 1.2. No native dependencies, no protobuf library (a minimal wire-format decoder is built in). Serve it with npx mockingbird-otel serve, createServer from ./server (Node), or createRuntime with any Fetch server.

Usage

Point OTEL_EXPORTER_OTLP_ENDPOINT at the mock and set OTEL_TRACES_SAMPLER_ARG=1 (our SDK samples 10% of root spans otherwise). Point O2_BASE_URL at the same mock; O2_BASIC_AUTH is any base64 user:password unless --search-auth is set.

npx mockingbird-otel serve --port 8809 --ingest-token "$OTEL_AUTH_TOKEN"

import { createRuntime } from "@crvouga/mockingbird-service-otel"

const otel = createRuntime()
const call = (path: string, init: RequestInit) => otel.fetch(new Request(`http://otel.test${path}`, init))

// What @opentelemetry/exporter-logs-otlp-http sends for one winston/pino event log.
await call("/v1/logs", {
  method: "POST",
  headers: { "content-type": "application/json", authorization: "Bearer ingest-token" },
  body: JSON.stringify({
    resourceLogs: [{
      resource: { attributes: [{ key: "service.name", value: { stringValue: "backend" } }] },
      scopeLogs: [{ logRecords: [{
        severityText: "info",
        body: { stringValue: "reconcile done" }, // dropped: bodies are never stored
        attributes: [
          { key: "event", value: { stringValue: "initial_credit_reconcile_completed" } },
          { key: "granted", value: { intValue: "1" } },
        ],
      }] }],
    }],
  }),
})

// Long-poll until it lands, then query it the way our O2 clients do.
await call("/__admin/wait", {
  method: "POST",
  headers: { "content-type": "application/json" },
  body: JSON.stringify({ kind: "log", where: { event: "initial_credit_reconcile_completed" } }),
})
const search = await call("/api/30rBqcDevOrg7Hn2KmQ4xW9sLtY/_search", {
  method: "POST",
  headers: { "content-type": "application/json", authorization: `Basic ${btoa("agent:pw")}` },
  body: JSON.stringify({
    query: { sql: `SELECT * FROM "default" WHERE event = 'initial_credit_reconcile_completed'` },
  }),
})
// { hits: [{ service_name: "backend", event: "…", granted: 1, severity_text: "info", _timestamp }], … }

Receiver

| Route | Behaviour | | --- | --- | | POST /v1/traces | ExportTraceServiceRequest, application/json or application/x-protobuf (optionally content-encoding: gzip) → 200 {partialSuccess: {}} (an empty protobuf response for protobuf requests). | | POST /v1/logs | ExportLogsServiceRequest, same encodings and answers. | | POST /v1/metrics | Accepted and counted (GET /__admin/otlp-metrics), never stored. |

Authorization: Bearer <token> is required (401 without); with --ingest-token only that token is accepted. Undecodable payloads are 400, other content types 415. The optional stream-name header picks the stream (default default).

Storage, as O2 stores it. Every field name is lowercased and flattened (clientUserId → clientuserid, http.status_code → http_status_code, nested maps joined with _); resource attributes get a service_ prefix (service_service_version, service_deployment_environment_name) except service.name → service_name; null fields are dropped. Logs carry _timestamp (µs), severity_text, severity_number, trace_id, span_id and their attributes (event, …). Spans carry operation_name, trace_id, span_id, reference_parent_span_id, span_kind, span_status (UNSET/OK/ERROR), start_time / end_time (ns), duration (µs), events (JSON) and their attributes. Log bodies are dropped unless --keep-bodies (a body can hold a prompt or PHI; truncation is not protection); the body column still exists in the schema so recipes that select it keep working.

Org routing. deployment.environment.name (or deployment.environment) production lands in the production org; anything else in development, like the collector's dev-org default. Default orgs: default, development (30rBqcDevOrg7Hn2KmQ4xW9sLtY), production (3HSzeProdOrg5Jd8VpN1cR6gTfB).

O2 search

| Route | Behaviour | | --- | --- | | GET /api/organizations | {data: [{id, identifier, name, type}]}: display name → identifier. | | GET /api/{org}/streams?type=logs\|traces | {list: [{name, stream_type, stats: {doc_num}}]}. | | GET /api/{org}/streams/{stream}/schema?type=logs\|traces | {name, stream_type, schema: [{name, type}]}: every field ever ingested non-null. 404 before anything is ingested. | | POST /api/{org}/_search?type=logs\|traces | {query: {sql, start_time, end_time (µs), from, size}} → {took, hits, total, from, size, scan_records, is_partial}. |

Basic auth is required (401 Unauthorized Access, text/plain). {org} must be an org identifier: a display name (/api/production/_search, as release-conductor sends) is the same bare 401 real O2 answers, so that bug stays visible.

The SQL subset: SELECT * | expr [AS alias], … FROM "<stream>" [WHERE …] [GROUP BY …] [HAVING …] [ORDER BY expr [ASC|DESC], …] [LIMIT n [OFFSET m]] with AND/OR/NOT, =, !=/<>, <, <=, >, >=, IS [NOT] NULL, [NOT] IN, [NOT] LIKE/ILIKE, [NOT] BETWEEN (ISO strings compare against _timestamp), and str_match, str_match_ignore_case, match_all, re_match, lower, upper, tostring, length, coalesce, count(*), count([DISTINCT] x), min, max, sum, avg. Rows outside start_time–end_time are excluded, the default order is _timestamp DESC, LIMIT applies before from/size paging. Column names are case-sensitive and must exist in the stream's schema: clientUserId or a never-ingested column is 400 … No field named …, as our clients expect. A query on a stream with no data answers empty hits.

Admin (beyond the standard contract)

| Route | Effect | | --- | --- | | GET /__admin/logs?service=&event=&severity=&trace_id=&org= | Stored log rows (with _org, _stream). | | GET /__admin/spans?service=&name=&trace_id=&org= | Stored span rows. | | POST /__admin/wait | {kind: "log"\|"span", where: {event: "…", …}, count?: 1, timeoutMs?: 5000, org?}: long-polls until count rows match (200 {matched, count}) or times out (408 with what matched). where keys may be spelt as emitted (clientUserId, service.name) or as stored. | | GET /__admin/otlp-metrics | {requests, bytes, metrics} for /v1/metrics. | | PUT /__admin/settings | {ingestTokens?, searchUsers?: [{username, password}], organizations?: [{identifier, name}], routing?: {byEnvironment, default}, keepBodies?} for the calling namespace. |

Fault presets (POST /__admin/faults {"preset": "<name>", "count"?: n}): rate_limited (429 + retry-after: 1), bad_gateway (502), unavailable (503), gateway_timeout (504) — all retried by the SDK — server_error (500) and unauthorized (401), which the SDK drops, partial_success (200 rejecting every item, JSON only) and search_unavailable (O2 search 503).

Namespaces

x-mockingbird-namespace, a /ns/<name> prefix on the endpoint and base URL, or by credential: PUT /__admin/credentials {"credentials": {"<OTEL_AUTH_TOKEN>": "w1", "<O2 username>": "w1"}} (map both so a worker's exports and searches meet).

Deliberately not modelled

• gRPC OTLP (port 4317); only OTLP/HTTP.
• Metrics storage and the O2 metrics/PromQL APIs: metric exports are counted only.
• O2's full SQL dialect (DataFusion): joins, subqueries, window functions, arithmetic, histogram(), approx_*; only the subset above.
• O2 ingestion via /api/{org}/{stream}/_json or /api/{org}/v1/logs, dashboards, alerts and the UI; the collector's own batching, tail sampling and transform processors.
• partialSuccess for protobuf requests (an empty success response is sent).
• Log bodies by default (see --keep-bodies).

API

| Export | Kind | Description | | --- | --- | --- | | OtelAPI | class | The in-process mock: fetch(request), reset(), logs(), spans(), state. Options: sqlite, now, namespace, settings. | | createRuntime | function | The mock with the full service contract (health, admin, wait, namespaces, credentials, presets). Options: settings, clock, seed, adminKey, onLog, sqlite. | | OTEL_PRESETS | object | Every named fault preset. | | OTEL_NAMESPACE | string | The service name, "otel". | | otelCredential | function | The OTLP bearer token or the O2 Basic username (how credentials map to namespaces). | | adminRow | function | A stored row as the admin routes show it (_org, _stream plus its columns). | | DEFAULT_ORGANIZATIONS, DEFAULT_SETTINGS | values | The seeded orgs and settings. | | logRows, spanRows, formatKey | functions | OTLP/JSON export → O2 rows, and O2's field-name normalisation. | | decodeLogsRequest, decodeTraceRequest, ProtobufError | functions, class | The protobuf decoder: OTLP protobuf → OTLP/JSON shape. | | parseSql, execute, referencedColumns, SqlError | functions, class | The SQL subset: parse, run over rows, list the columns a query reads. | | document, operationIds, supportedOperationIds | values | The OpenAPI contract and its operation ids. | | createServer, serveTarget, DEFAULT_PORT (./server) | Node | Serve over node:http; the serve CLI target; port 8809. |

Part of mockingbird.