Sinks¶

A sink is where events go. The context holds exactly one, as Arc<dyn TelemetrySink>.

The TelemetrySink trait¶

#[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.

Which sink to use¶

"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.

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.

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.

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¶

Constructors¶

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:

HTTP telemetry sink error: endpoint scheme "http" not permitted (set allow_insecure_endpoint for tests)

A transport failure surfaces the underlying reqwest message:

HTTP telemetry sink error: error sending request for url (https://127.0.0.1:9/x)

A non-2xx response is an error too:

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¶

How the transport is chosen¶

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:

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:

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:

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
• 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.