---
title: Sinks
description: The TelemetrySink trait and the five built-in sinks, with every configuration field, default and failure mode.
---
# Sinks
A sink is where events go. The context holds exactly one, as
`Arc<dyn TelemetrySink>`.
## The `TelemetrySink` trait
```rust
#[async_trait]
pub trait TelemetrySink: Send + Sync + 'static {
async fn emit(&self, event: &Event) -> Result<(), TelemetryError>;
async fn flush(&self) -> Result<(), TelemetryError> { Ok(()) }
}
```
`emit` takes `&self`, so a sink handles its own interior mutability.
`flush` has a default no-op implementation; override it only if the sink
buffers. There is no `shutdown` or `Drop` contract — a sink that must
finish work before the process exits has to expose that itself, and the
tool has to call it.
Writing your own is a short job: see
[Write a custom sink](../how-to/write-a-custom-sink.md).
## Which sink to use
| `NoopSink` | always | nowhere | n/a |
| `MemorySink` | always | a `Vec<Event>` in this process | **no** |
| `FileSink` | always | newline-delimited JSON on disk | yes |
| `HttpSink` | `remote-sinks` | JSON POST to an HTTPS endpoint | yes |
| `OtlpSink` | `remote-sinks` | OTLP collector, as log records | yes |
"Redacts" means the sink calls `Event::redacted()`, which passes `args` and
`err_msg` through `rtb-redact`. It never touches `attrs`. The two sinks
that do not redact are the two that never leave the process.
## `NoopSink`
A unit struct that returns `Ok(())` from `emit` and drops the event. It is
the builder's default sink, and it is a legitimate production
configuration: a tool can ship the whole telemetry surface — the consent
commands, the recording call sites — while sending data precisely nowhere.
## `MemorySink`
An in-process `Arc<Mutex<Vec<Event>>>`, intended for tests.
| `MemorySink::new()` | empty sink |
| `.snapshot()` | `Vec<Event>` — a clone of everything recorded, in emit order |
| `.len()` | count of recorded events |
| `.is_empty()` | `true` when nothing has been recorded |
`Clone` shares the same backing vector, so cloning the sink into a context
and keeping a handle for assertions works as expected.
Two things to know before relying on it outside tests. It stores the event
**unredacted**, exactly as the caller built it. And if the internal mutex
has been poisoned by a panic in another thread, `emit` silently drops the
event and still returns `Ok(())`, while `snapshot()` returns an empty
vector — a poisoned `MemorySink` looks like a sink that was never written
to.
## `FileSink`
Appends each event as one line of JSON to a file — JSON Lines.
```rust
FileSink::new(path) // path: impl Into<PathBuf>
```
- The file is not touched until the first `emit`. Parent directories are
created then, not at construction.
- The file is opened with `create(true).append(true)` on every `emit`, so
an existing file is appended to and never truncated.
- Each line is the **redacted** event, so a credential that reached
`args` or `err_msg` does not reach the disk.
- Concurrent `emit` calls are serialised through a shared
`tokio::sync::Mutex`, which is what keeps two large events from
interleaving into one malformed line. The reasoning is in
[Why FileSink holds a lock](../explanation/file-sink-line-integrity.md).
What it does not do: rotate, cap, compress, or delete the file. It grows
until something else removes it. And the lock is process-local — two
processes appending to one path are back to relying on the operating
system, which only guarantees non-interleaving below 4 KiB per write on
Linux. Give each process its own file.
Failures surface as `TelemetryError::Io` (directory creation, open, write,
flush) or `TelemetryError::Serde` (the event could not be serialised).
## `HttpSink`
POSTs one JSON body per event to a configured endpoint. Requires the
`remote-sinks` feature.
### `HttpSinkConfig` fields
| `endpoint` | `url::Url` | `https://telemetry.invalid/` | Full URL including path. The default is a deliberate placeholder — override it. |
| `bearer_token` | `Option<SecretString>` | `None` | Sent as `Authorization: Bearer <token>`. `Debug` prints `[REDACTED]` and the memory is zeroed on drop. |
| `timeout` | `Duration` | 5s | Per-request timeout. **Ignored** when the client is injected with `with_client`. |
| `user_agent` | `String` | `"rtb-telemetry/0.2"` | Sent as `User-Agent`. **Ignored** when the client is injected. |
| `allow_insecure_endpoint` | `bool` | `false` | When `true`, `http://` endpoints are accepted. For tests against a local mock server. |
### Constructors
```rust
HttpSink::new(config) // builds its own reqwest::Client — Result
HttpSink::with_client(config, client) // uses yours — infallible
```
`new` validates the endpoint scheme up front and returns
`TelemetryError::Http` if it fails. `with_client` cannot fail at
construction, so the scheme is re-checked on every `emit` instead: a
misconfigured endpoint surfaces the same error either way, just later.
### Errors
Anything other than an HTTPS endpoint (or `http://` with
`allow_insecure_endpoint`) is refused:
```text
HTTP telemetry sink error: endpoint scheme "http" not permitted (set allow_insecure_endpoint for tests)
```
A transport failure surfaces the underlying reqwest message:
```text
HTTP telemetry sink error: error sending request for url (https://127.0.0.1:9/x)
```
A non-2xx response is an error too:
```text
HTTP telemetry sink error: non-2xx response: 404 Not Found
```
### Limits
One POST per event, synchronously awaited. No batching, no retry, no
queue, no back-off: a slow collector slows down whatever awaited `record`,
bounded by `timeout`. If that matters for your tool, wrap the sink in
something that buffers, or write to a `FileSink` and ship the file.
The POST body is the redacted event plus a top-level `severity` field.
## `OtlpSink`
Exports each event as one OpenTelemetry log record. Requires the
`remote-sinks` feature.
### `OtlpSinkConfig` fields
| `endpoint` | `String` | `"http://127.0.0.1:4317"` | Collector endpoint. The scheme and port pick the transport — see below. |
| `headers` | `Vec<(String, SecretString)>` | empty | Sent as gRPC metadata or HTTP headers. |
| `timeout` | `Duration` | 10s | Per-export timeout. |
| `resource_attrs` | `Vec<(String, String)>` | empty | Extra OpenTelemetry resource attributes. |
### How the transport is chosen
| `grpc://…`, `grpcs://…` | OTLP/gRPC via tonic |
| any `http(s)://` URL containing `:4317` | OTLP/gRPC via tonic |
| any other `http(s)://` URL | OTLP/HTTP with a protobuf payload |
| anything else | rejected at construction |
The `:4317` rule is a substring test on the whole endpoint string, not a
port parse. It exists because a collector on the OTLP gRPC default port
almost always speaks gRPC even when the URL was typed with an `http://`
scheme. The consequence is that you cannot select the HTTP/protobuf
transport for an endpoint whose URL contains `:4317` anywhere, including
in a path.
Construction rejects an unrecognised scheme and an empty gRPC host:
```text
OTLP telemetry sink error: unsupported endpoint scheme in "tcp://nope" (expected grpc://, grpcs://, http://, or https://)
OTLP telemetry sink error: empty host in endpoint "grpc://"
```
### `emit` never reports an export failure
`OtlpSink::emit` hands the record to the SDK's batch processor and returns
`Ok(())`. It returns `Ok(())` when the collector is refusing connections,
when the endpoint is wrong, and when the export later fails. The only place
an export error surfaces is `flush`:
```text
OTLP telemetry sink error: flush: Operation failed: errs: [Err(InternalFailure("TonicLogsClient export failed with gRPC code: Unavailable: transport error: tcp connect error: tcp connect error: Connection refused (os error 111)"))]
```
So a tool that wants to know whether OTLP export worked has to call
`TelemetryContext::flush()` and check its result. Nothing else will tell
it.
### The OTLP/HTTP transport does not work in 0.7.3
Tested against rtb-telemetry 0.7.3 with opentelemetry 0.32.0,
opentelemetry_sdk 0.32.1 and opentelemetry-otlp 0.32.0: an endpoint routed
to the HTTP/protobuf transport (any `http(s)://` URL without `:4317`, for
example `http://127.0.0.1:4318/v1/logs`) panics inside the SDK's exporter
thread the moment it tries to send, and exports nothing:
```text
thread 'OpenTelemetry.Logs.BatchProcessor' panicked at …:
there is no reactor running, must be called from the context of a Tokio 1.x runtime
```
`emit` still returns `Ok(())`; the subsequent `flush` returns
`OTLP telemetry sink error: flush: … InternalFailure("channel is empty and sending half is closed")`.
Use `grpc://`, `grpcs://`, or an endpoint on `:4317` until this is fixed.
Most collectors accept OTLP/gRPC on 4317 by default, so this is usually a
one-word change to the endpoint.
### What the collector sees
Each event becomes one log record with:
- event name `rtb.telemetry.event`
- severity number and text from the [severity rule](event.md#severity)
- the body: the redacted event serialised as a JSON string
- attributes `tool`, `tool.version` and `event.name`
The resource carries `service.name = "rtb-telemetry"` — the constant, not
your tool's name — the SDK's own `telemetry.sdk.*` attributes, and whatever
you put in `resource_attrs`. Your tool's identity travels as the `tool` and
`tool.version` record attributes, not as the service name, so anything that
groups by `service.name` groups every rtb-telemetry producer together.
`resource_attrs` entries are applied after the service name, so passing
`("service.name", "mytool")` overrides the constant (verified against
opentelemetry_sdk 0.32.1). Do that if your backend's navigation is built
around services.