Emitters

OTLP over HTTP with SigV4 and protobuf — the three endpoints, the per-request limits, on-demand log streams, and the RUM data plane.

An emitter takes rendered records and puts them somewhere. Real emitters talk to AWS; RecordingEmitter keeps them in memory, which is what --dry-run, the cost estimator and every test use. Nothing above this layer knows which it has.

The Emitter protocol#

proteus/emit/base.py is frozen.

@runtime_checkable
class Emitter(Protocol):
    signal: Signal
    def send(self, records: Sequence[Record]) -> EmitResult: ...

Two obligations on every implementation:

• Split your own input into wire-legal batches. Callers hand over arbitrarily many records.
• Never raise on a partial failure. Record it in EmitResult.errors and carry on, so one bad batch cannot abort a 14-day backfill.

@dataclass(slots=True)
class EmitResult:
    accepted: int = 0
    rejected: int = 0
    requests: int = 0
    bytes_sent: int = 0
    throttled: int = 0          # requests that came back 429
    errors: list[str] = field(default_factory=list)

    def merge(self, other) -> EmitResult: ...   # neither input is mutated
    @property
    def ok(self) -> bool: ...                   # nothing rejected, nothing errored

EmitterSet fans a mixed record stream out per signal using records.signal_of. Records whose signal has no registered emitter are counted as rejected under that signal rather than silently dropped.

The three OTLP endpoints#

HTTP only. There is no gRPC endpoint. Bodies are OTLP 1.x protobuf (application/x-protobuf) with all timestamps in nanoseconds since the Unix epoch.

def endpoint_for(signal: Signal, region: str) -> str:
    spec = SPECS[signal]
    return f"https://{spec.service}.{region}.amazonaws.com{spec.path}"

Logs additionally require the x-aws-log-group and x-aws-log-stream headers. Traces require Transaction Search to be enabled, with the destination set correctly; see the two AWS-side prerequisites.

emit/otlp.py::SPECS carries the hard per-request limits, taken verbatim from CONTRACTS.md:

The 1,000-datapoint metrics cap is the tightest limit in the system and binds long before the byte one.

Logs are capped at 1 MB in most regions and 20 MB in a handful (us-east-1, us-west-2, eu-west-1 "and others"). Because the full list is not documented, Proteus uses the conservative 1 MB everywhere — and the record cap binds first anyway.

_OtlpEmitter#

The three signal emitters are thin subclasses of one base. Subclasses supply the signal, the record type they accept, and how to serialise a batch; everything else — batching, window validation, signing, retry, dry run and result accounting — lives in _OtlpEmitter.

class TraceEmitter(_OtlpEmitter):
    signal = Signal.TRACES
    _record_type = SpanRecord
    def _build(self, records): return _proto.build_trace_request(records)
    def _timestamp_of(self, record): return record.start

MetricEmitter and LogEmitter are the same shape, using record.timestamp. LogEmitter additionally overrides _header_groups and _prepare_group.

send() runs four steps, and each can shed records without raising:

sequenceDiagram
    participant C as caller
    participant E as _OtlpEmitter
    participant P as SignedPoster
    participant AWS as CloudWatch

    C->>E: send(records)
    E->>E: _screen: wrong type, outside window
    Note over E: rejected, counted, never sent
    E->>E: _header_groups: split by shared headers
    E->>E: _prepare_group: create log stream
    Note over E: a failure here rejects the group
    E->>E: plan_batches: cut to record and byte caps
    loop each batch
        E->>P: post(url, body, headers)
        P->>P: gzip at 1 KiB or above, then SigV4 sign
        P->>AWS: POST
        AWS-->>P: status
        P->>P: retry with jitter on 408, 429, 5xx
        P-->>E: HttpOutcome
    end
    E-->>C: EmitResult

1. _screen splits input into sendable records and an EmitResult of rejections.
2. _header_groups partitions the survivors into groups that may share headers.
3. _prepare_group does any per-group setup (this is where log streams get created).
4. _proto.plan_batches cuts each group into wire-legal batches, and each goes out through SignedPoster.

Two properties of that sequence are worth reading off it. Nothing between send() and the endpoint raises on an HTTP or transport failure — every arm returns an outcome that merges into one EmitResult, so a partial failure is reported rather than thrown. And gzip happens before signing, because SigV4 signs the bytes actually on the wire; a body compressed after signing would fail authentication rather than fail validation.

_screen rejects rather than sending a doomed request#

Two classes of record are counted as rejected with one aggregated error each, rather than being put into a request that is certain to fail:

• Wrong type — anything that is not this emitter's _record_type.
• Outside the accepted timestamp window, when validate_window is on (the default). The bounds come from the signal's past_seconds and future_seconds.

metrics: dropped 412 record(s) whose timestamp is outside the accepted window
(2026-09-11T09:14:00+00:00 .. 2026-09-25T09:24:00+00:00); CloudWatch would reject
the whole request. Re-render the window against the current clock.

_clock is injectable, so this is testable without waiting.

Dry run is real accounting#

if self.dry_run:
    return HttpOutcome(status=0, bytes_sent=len(wire), attempts=0, dry_run=True)

dry_run=True builds and compresses every body, so bytes_sent and the batch count are real — but it opens no socket and resolves no credentials. That is what makes the paranoid guard policy an honest promise rather than a hope; see the cost guard.

Protobuf serialisation#

proteus/emit/_proto.py is private to proteus/emit/. Keeping the protobuf here means otlp.py stays readable and this file can be unit-tested by serialising and re-parsing without any AWS involvement.

Nanoseconds, computed with integers#

def unix_nanos(ts: datetime) -> int:
    if ts.tzinfo is None:
        ts = ts.replace(tzinfo=timezone.utc)
    delta = ts.astimezone(timezone.utc) - _EPOCH
    return (delta.days * 86_400 + delta.seconds) * 1_000_000_000 + delta.microseconds * 1_000

Not datetime.timestamp() * 1e9. That loses roughly a hundred nanoseconds of precision at current epoch values, and the error is invisible until a demo shows spans in the wrong order.

Byte-identical output#

attributes_to_kv emits keys in sorted order, so identical input serialises to identical bytes. The golden-file tests depend on it.

_set_any_value checks bool first, because bool is a subclass of int in Python and True must not become an int_value.

hex_to_bytes(value, width=...) converts a lowercase hex id to raw bytes: 16 bytes for a trace id, 8 for a span id. A short id is left-padded and an over-long one is truncated from the left, so a malformed id degrades instead of aborting a backfill.

Grouping#

Records are grouped into one ResourceSpans / ResourceMetrics / ResourceLogs per distinct Resource.key(), preserving first-seen order. Metric points are then grouped again into one Metric per (name, kind, unit, is_monotonic) so a series carries all of its datapoints together.

SUM and HISTOGRAM are marked AGGREGATION_TEMPORALITY_DELTA, and delta points carry start_time_unix_nano == time_unix_nano because a MetricPoint records an instant, not an interval.

plan_batches#

def plan_batches(records, *, build, timestamp_of,
                 max_count, max_bytes, max_span_seconds=None) -> list[Batch]: ...

Enforces the three limits in order: record count, timestamp gap, then byte size. The byte limit is enforced by serialising and bisecting — _fit builds the body, and if it is over the ceiling it splits the chunk in half and re-serialises each half. That is exact rather than estimated, which is why this exists alongside render/timeline.py::estimate_bytes rather than reusing it.

A single record that is still oversized alone is returned with oversized=True, and the emitter rejects it with an actionable message rather than sending a request that is certain to fail:

traces: a single record serialises to 231884 bytes, over the 5242880-byte
per-request ceiling; not sent

Input order is preserved throughout.

opentelemetry-proto is imported through a functools.cached pb() helper that raises ImportError naming the install command rather than letting a bare No module named opentelemetry.proto escape. proto_available() checks without raising.

SigV4, gzip, retry and throttling#

proteus/emit/_transport.py moves bytes and reports what happened. Nothing in it knows about OTLP or RUM payloads.

Two rules it exists to enforce:

• Nothing leaves the process when dry_run is set. The body is still built and measured, so size accounting is real, but no socket opens and no credentials are resolved.
• Failures are returned, never raised. An HttpOutcome with status == 0 means the request never got a response at all.

Credentials are never logged, echoed into an error string, or put in an HttpOutcome.

Signing#

request = AWSRequest(method="POST", url=url, data=body, headers=dict(headers))
SigV4Auth(self._frozen_credentials(), self.service, self.region).add_auth(request)

The body is signed exactly as it will go on the wire, i.e. after gzip. Credentials are resolved once and frozen; _frozen_credentials raises NoCredentialsError if the session has none, and is never called in dry run.

gzip#

GZIP_MIN_BYTES = 1024

Bodies at least that large are gzipped before signing; below it the compression overhead outweighs the saving. gzip_body zeroes mtime so the output is reproducible.

Retry#

RETRYABLE_STATUSES = frozenset({408, 429, 500, 502, 503, 504, 509})

Plus status 0, which means no response arrived at all. Default max_attempts is 4.

def backoff_delay(attempt, rng, *, base=0.5, cap=8.0) -> float:
    ceiling = min(cap, base * (2 ** max(0, attempt - 1)))
    return ceiling * (0.5 + rng.random())

Exponential, capped at 8 seconds, multiplied by jitter in [0.5, 1.5) drawn from an injected rng. Each emitter seeds its own from _default_seed(signal), so a seeded emitter replays identical delays in a test.

Self-throttling#

TokenBucket keeps a process under the account-wide requests-per-second ceiling — 500 for metrics, 5,000 for logs, none published for traces. acquire() blocks until a token is available and returns the seconds it slept, which SignedPoster.seconds_throttled accumulates, so self-throttling is reportable rather than invisible. clock and sleeper are injectable.

The practical consequence, surfaced by the cost guard as a warning: a run that projects past the TPS cap does not fail, it just takes longer than the window it covers.

Log streams are created on demand#

The /v1/logs endpoint requires that both the log group and the log stream already exist. The headers auto-create neither: a missing stream fails the whole batch with HTTP 400 "The specified log stream does not exist."

The split of responsibility:

• Log groups are the CloudFormation stack's job. LogStreamEnsurer never creates one — quietly creating an untracked group would break stack teardown.
• Log streams are the emitter's job. Their names are generated per service instance by the catalog at render time, so there is nothing useful for CloudFormation to declare.

emit/_logstreams.py::LogStreamEnsurer.ensure(group, stream) calls logs:CreateLogStream lazily, once per pair, caching what it has ensured so a 14-day backfill makes a handful of calls rather than thousands. It returns None on success or an actionable string on failure, and never raises.

region_name is passed explicitly when building the client rather than trusting the session's ambient region — an unpinned session resolves to whatever the environment says, which is how streams get created in one region while telemetry goes to another.

Error handling by code:

OperationAbortedException is in the retry set because it means a concurrent create raced us.

LogEmitter exposes ensured_streams and create_log_stream_calls for inspection, and create_streams=False turns the whole mechanism off for a caller that pre-creates streams itself.

Why the log-group set is derived, never sampled#

This is the most important operational detail in the emitter layer, and it is a bootstrap/doctor concern rather than a send one.

A log group that does not exist fails every batch for that service with HTTP 400. So proteus bootstrap has to provision exactly the set the renderer can write to, and proteus doctor has to check the deployment against that set.

The original implementation sampled a short render and collected the log groups it saw. That was wrong, and measurably so:

• A 5-minute probe is 5 ticks. An undriven log line at frequency 0.12 misses all five about 53% of the time, so one service's log group was being created by coin flip.
• Two services whose only declared log lines were driver-gated — emitted only while a fault is active — got no group at all.
• Measured afterwards: all 13 applicable fault kinds make each of those services emit 58–120 log records per 30 minutes, every one of which would have been rejected.

A driver-gated line is invisible to any healthy probe regardless of window width. Widening the window does not fix a sampling approach; it only makes the coin fairer.

proteus/cli/doctor.py::expected_log_groups therefore unions two sources, because neither alone is sufficient:

groups = {f"/proteus/{model.name}/{service.name}" for service in model.services}

now = utc_now()
probe = Window(start=now - timedelta(hours=2), end=now, interval_s=120.0)
groups |= {
    record.log_group
    for record in render(model, probe)
    if isinstance(record, LogRecord)
}

The convention, derived per service. /proteus/{model}/{service} for every service in the model. This has to be derived rather than observed, because of the driver-gated case above — and because every service additionally emits fault_logs during any fault regardless of what it declares. A service can be silent today and log heavily the moment a fault lands on it.

A rendered probe, as a safety net. For an archetype using some scheme other than the convention. An earlier catalog emitted realistic names like /aws/bedrock/agents/..., which is how that drift was first found. Rendering keeps such a scheme impossible to miss silently.

Correctness now rests on the derivation. The probe only catches drift.

doctor reports a missing group as a FAIL, not a warning:

log groups   FAIL   2 log group(s) the renderer can write to do not exist:
                    /proteus/northwind-rx/orders-queue,
                    /proteus/northwind-rx/rx-inference. Every log batch for those
                    services will be rejected with HTTP 400 - including services
                    that only log while a fault is active, which is when you most
                    need them.
             fix:   proteus bootstrap   (re-applies the stack)

Log retention is a data-loss trap#

CloudWatch Logs applies retention against the event timestamp, not the ingest time. A backdated event older than the group's retention period is accepted with HTTP 200 and then silently discarded. It is never retrievable and nothing reports an error.

Verified: events at −25 h and −10 d into a 1-day-retention group returned 200 and were never retrievable.

Two defences, in different layers:

• cfn/synth.py hard refuses to synthesise a template whose retention is shorter than the declared seed horizon. See provisioning.
• proteus doctor compares the deployed retention against the backfill window and reports a FAIL.

The default retention is 14 days, matching the OTLP −14 d limit. It is not a cost-saving knob.

Diagnosing a rejection#

A silent rejection is the worst outcome for a demo, so emit/_diagnose.py turns a status and response body into a sentence that names a concrete next action. diagnose is re-exported from proteus.emit.otlp, which is where callers should import it from, and _OtlpEmitter._send_batch appends its output to the error:

logs: batch of 1204 failed with HTTP 400: ... | likely cause: The CloudWatch Logs
OTLP endpoint rejected this batch because the log group does not exist. Log groups
are created by the CloudFormation stack, not by the emitter. Run 'proteus
bootstrap' to create them, then retry.

It recognises, in order: Transaction Search / trace segment destination; a missing log group (checked before a missing stream, because a missing group is a bootstrap problem and a missing stream is ours to create); expired credentials; SigV4 signature rejection; access denied; a timestamp outside the window; throttling; a reserved AWS/ namespace; 404; and 413.

Full symptom-to-fix coverage is in Troubleshooting.

The two AWS-side prerequisites#

Transaction Search must have destination CloudWatchLogs, not XRay. It is not enough for it to be active. With destination XRay, every span is rejected with HTTP 400 naming UpdateTraceSegmentDestination. It also needs a CloudWatch Logs resource policy letting xray.amazonaws.com write to aws/spans.

doctor checks both the destination and the status, and distinguishes them:

transaction search   FAIL   destination is 'XRay' (status ACTIVE). The OTLP traces
                            endpoint requires CloudWatchLogs; every span will be
                            rejected with HTTP 400 until this changes.
                     fix:   proteus bootstrap   (or: aws logs put-resource-policy
                            granting xray.amazonaws.com logs:PutLogEvents on
                            aws/spans, then aws xray
                            update-trace-segment-destination --destination
                            CloudWatchLogs)

Destination CloudWatchLogs with a non-ACTIVE status is a WARN, not a FAIL: spans may still be rejected for a few minutes while it settles, so the fix is "re-run doctor shortly".

No namespace beginning AWS/ may be published to. Reserved by AWS; the request is rejected. Real AWS metric names travel as the attributes proteus.aws_metric and proteus.aws_namespace, never as the emitted namespace. See the catalog.

RUM is a different shape#

proteus/emit/rum.py sends RumEvents to a CloudWatch RUM app monitor's data plane via PutRumEvents, at:

https://dataplane.rum.{region}.amazonaws.com/appmonitors/{Id}/

The trailing slash is part of the modelled request URI and is required.

It prefers the boto3 rum client — whose service model carries hostPrefix: dataplane., so it targets that URL by itself — and falls back to signed HTTP against the same URL when no client is available. gzip is deliberately off: the RUM data plane's acceptance of a compressed body is not documented, and a RUM batch is small.

The request shape is verified against the installed botocore service model, not guessed. PutRumEvents requires Id, BatchId, AppMonitorDetails, UserDetails and RumEvents. Three consequences:

• Id, BatchId, UserDetails.sessionId, UserDetails.userId and every RumEvents[].id have a minimum length of 36 characters — they are UUIDs. stable_uuid(*parts) maps a human-readable RumEvent id onto a deterministic UUID v5 under a fixed namespace, so a re-run of the same model produces the same ids.
• metadata and details are JSON strings, not objects.
• Timestamps serialise as whole Unix seconds, so sub-second ordering within a RUM batch is not preserved on the wire.

Because UserDetails is per request, records are grouped by (session_id, user_id) and two sessions never share a request — the same constraint shape as the log group/stream headers.

A monitor id shorter than 36 characters is rejected up front with everything unsent, because the common mistake is passing the monitor's name:

rum: app monitor id 'northwind-storefront' is 20 characters; PutRumEvents requires
the monitor's 36-character UUID id (not its name). Nothing sent.

proteus bootstrap creates the monitor and the stack exposes its id as the RumAppMonitorId output precisely so the CLI has the UUID rather than the name.

The per-request caps are self-imposed and need verification: AWS publishes no per-request event or byte ceiling for PutRumEvents, so DEFAULT_MAX_EVENTS_PER_REQUEST = 200 and DEFAULT_MAX_BYTES_PER_REQUEST = 512 * 1024 are deliberately conservative guesses. Check them against the RUM service quotas before raising them.

When the boto3 client is used, botocore's own retries are turned off (max_attempts=1) so this class owns the retry loop and its jitter stays seeded and repeatable.