Skip to content

Sinks

Sinks are the terminal points of a pipeline that consume or export data.

Debug Sink

Writes every event to the log on stderr as a Debug event record with the attributes path (the path the event took through the stream, e.g. in -> filter -> out), records (the number of log records, data points and spans) and event (Go's representation of the OTLP data, which is verbose but complete). Useful for development, testing, and debugging pipelines.

Type: debug

Configuration: None required.

Example:

{
  "name": "console-output",
  "type": "debug",
  "from": "source"
}

OTLP Sink

Sends OpenTelemetry Protocol data via HTTP to an OTLP-compatible endpoint. Sends logs to /v1/logs, metrics to /v1/metrics, and traces to /v1/traces with a protobuf-encoded body.

Type: otlp

Configuration:

Field Type Required Default Description
url string Without url_expression - OTLP HTTP endpoint URL (e.g., "http://localhost:4318")
url_expression object No - Sends each entry to the URL a CEL expression returns, see Routing by URL
headers object No - HTTP headers to send with each request
timeout string No "10s" HTTP request timeout as a Go duration (e.g., "5s", "30s")
insecure boolean No false Allow connections to servers with invalid TLS certificates

Example:

{
  "name": "otlp-sink",
  "type": "otlp",
  "from": "source",
  "spec": {
    "url": "http://telemetry-collector:4318"
  }
}

Example with authentication and custom timeout:

{
  "name": "secure-otlp-sink",
  "type": "otlp",
  "from": "filter-stage",
  "spec": {
    "url": "https://telemetry-collector:4318",
    "timeout": "30s",
    "headers": {
      "Authorization": "Bearer your-api-key-here"
    }
  }
}

Example with invalid certificate acceptance (for testing):

{
  "name": "dev-otlp-sink",
  "type": "otlp",
  "from": "source",
  "spec": {
    "url": "https://localhost:4318",
    "insecure": true
  }
}

Behavior Notes: - Each signal type of an event is sent as a separate HTTP request to its endpoint. An event can hold several signal types, e.g. after a batch stage; if one request fails, the source retries the whole event, so the others may be sent twice - Uses protobuf encoding (application/x-protobuf content type) as required by OTLP/HTTP - The sink handles URL construction properly, so both "http://host:4318" and "http://host:4318/" work correctly - Non-2xx answers fail the delivery. Client errors that refuse the data itself (4xx except 408 and 429) reject the event, so the OTLP source answers with 422 and the NATS source terminates the message. Other errors, including 408, 429 and 5xx, can be retried by the source: the OTLP source answers with 503, the NATS source redelivers the message - Proxy settings are taken from the HTTPS_PROXY, HTTP_PROXY and NO_PROXY environment variables - Connection pooling is used for better performance with multiple requests - Timeout applies to the entire request lifecycle, including connection establishment and reading the response - The requests of an event are sent at the same time, at most 16 at once. If one fails in a way that may go away, e.g. with 503, the event fails retriably even if another destination rejected its data, so the source retries all of it. Only if every failed request was rejected is the event rejected

Routing by URL

With url_expression, the sink sends each log record, metric data point and span to the URL its expression returns, e.g. to a tenant's own endpoint. It groups the entries by URL and sends one request per URL and signal type.

Field Type Required Default Description
logs string One of logs, metrics, traces - CEL expression returning the URL of a log record
metrics string One of logs, metrics, traces - CEL expression returning the URL of a metric data point
traces string One of logs, metrics, traces - CEL expression returning the URL of a span
  • The expressions have the same variables and get_attr function as the subject expressions of the NATS Sink, e.g. record, resource and datapoint. Metrics are split per data point the same way
  • A URL is the base URL like url; the sink appends /v1/logs, /v1/metrics or /v1/traces. It must be http or https and have a host. An expression that fails, does not return a string, or returns another URL rejects the event. get_attr returns "" for a missing attribute, which is not a valid URL
  • Signals without an expression are sent to url. Without url, they are not sent
  • headers, timeout and insecure apply to every destination
  • The data decides where it is sent. Make sure only trusted values can reach the expression, e.g. by deriving the URL from a tenant ID that a cel_validator checks, rather than taking a URL from an attribute

Example with an endpoint per tenant:

{
  "name": "tenant-otlp",
  "type": "otlp",
  "from": "validator",
  "spec": {
    "url_expression": {
      "logs": "'https://' + record.attributes.get_attr('tenant.id') + '.otlp.example.com'"
    }
  }
}

NATS Sink

Publishes OTLP-encoded events to NATS JetStream. Uses dynamic subject mapping with CEL expressions to route different telemetry data to appropriate subjects. Creates the specified stream if it does not exist, and otherwise only adds what the sink needs to it.

Type: nats

Configuration:

Field Type Required Default Description
url string Yes - NATS server URL (e.g., "nats://localhost:4222")
stream string Yes - JetStream stream name to publish to
subject object Yes - Subject configuration for dynamic subject mapping
retention string No "limits" Stream message retention policy: "limits" (retain until limits are reached), "interest" (remove when all consumers acknowledge), or "workqueue" (remove when first consumer acknowledges)
max_age string No - Maximum age for messages in the stream as a Go duration (e.g., "24h", "168h"; days are not supported)
size int No - Maximum stream size in bytes
compression string No - How JetStream compresses the stream on disk: "s2" or "none"
replicas int No - How many NATS servers store the stream, from 1 to 5. A new stream without it gets the server's default, 1
discard string No "old" What JetStream does when the stream reaches size or another limit: "old" deletes the oldest messages, "new" refuses new ones
storage string No "file" Where JetStream stores the stream: "file" or "memory". Fixed once the stream exists
partition object No - Spreads records over several streams, see Partitioning

Subject Configuration:

Field Type Required Default Description
register array[string] Yes - List of subject patterns to register with the stream
logs string One of logs, metrics, traces - CEL expression to generate subject for log events. Must return a string. Has access to the logs variable
metrics string One of logs, metrics, traces - CEL expression to generate subject for metric events. Must return a string. Has access to the metrics variable
traces string One of logs, metrics, traces - CEL expression to generate subject for trace events. Must return a string. Has access to the traces variable

Signals without an expression are not published by this sink.

CEL Expression Context:

Expression Variables
logs logs (payload), resource (Resource), scope (InstrumentationScope), record (LogRecord)
metrics metrics (payload), resource, scope, metric (Metric), datapoint (the data point; its type depends on the metric type)
traces traces (payload), resource, scope, span (Span)

With partitioning, subject expressions also have partition (int), the partition of the record, data point or span.

  • Each expression is evaluated once per log record, metric data point and span. The payload variable holds only the current log record, metric or span with its resource and scope, so [0] indexes address it, e.g. logs.resource_logs[0].scope_logs[0].log_records[0]. For metrics it holds the whole metric with all its data points
  • resource and scope are empty messages when the payload has none. Field names follow the OTLP protobuf definitions, e.g. record.severity_text, datapoint.attributes
  • Metrics are routed per data point: a metric whose data points resolve to different subjects is split into one metric per subject, each with the metric's name, description, unit, metadata and type-specific fields like a sum's temporality. A metric whose data points all resolve to the same subject is published as it is, so an expression that does not use datapoint, e.g. metric.gauge.data_points[0].attributes.get_attr('team'), keeps every metric whole
  • A metric without data points is evaluated once without datapoint; an expression that reads it fails for that metric and rejects the event
  • The custom function get_attr(key) on an attribute list returns the string value of an attribute, or "" if it is missing, e.g. logs.resource_logs[0].resource.attributes.get_attr('service.name') or datapoint.attributes.get_attr('tenant.id')
  • Must return a string that will be used as the NATS subject

Example:

{
  "name": "nats_sink",
  "type": "nats",
  "from": "cel_stage",
  "spec": {
    "url": "nats://localhost:4222",
    "stream": "tram-stream",
    "subject": {
      "register": [
        "logs.audit.*",
        "metrics.*",
        "traces.*"
      ],
      "logs": "'logs.audit.' + logs.resource_logs[0].resource.attributes.get_attr('service.name')",
      "metrics": "'metrics.' + metrics.resource_metrics[0].resource.attributes.get_attr('service.name')",
      "traces": "'traces.' + traces.resource_spans[0].resource.attributes.get_attr('service.name')"
    }
  }
}

Example with retention and max_age limits:

{
  "name": "nats_sink_with_limits",
  "type": "nats",
  "from": "filter_stage",
  "spec": {
    "url": "nats://localhost:4222",
    "stream": "tram-data",
    "retention": "interest",
    "max_age": "24h",
    "size": 1073741824,
    "subject": {
      "register": ["logs.*", "metrics.*"],
      "logs": "'logs.' + logs.resource_logs[0].resource.attributes.get_attr('service.name')",
      "metrics": "'metrics.' + metrics.resource_metrics[0].resource.attributes.get_attr('service.name')"
    }
  }
}

Behavior Notes: - When the stream is created in Tram, the sink creates a missing JetStream stream from register, retention, max_age, size, compression, replicas, discard and storage. An existing stream is only extended: register subjects it does not cover yet are added, and retention, max_age, size, compression, replicas and discard are applied if set. Other settings, limits that are not set, and the subjects of other sinks or writers are kept, so several sinks can share a stream and it can be configured outside Tram - JetStream cannot change the storage of an existing stream. If storage is set and the stream has another one, the sink is not created and the Tram stream fails with an error, instead of the sink writing to a stream that does not keep its messages the way it asks for - replicas above 1 needs a NATS cluster with at least that many JetStream servers; otherwise creating or updating the stream fails. Changing it on an existing stream makes JetStream add or remove copies - Subjects are never removed from a stream; removing one from register leaves it on the stream. Set retention, size, compression, replicas and discard in one sink only, since sinks with different values replace each other's when they start - With discard: "old", a stream that reaches a limit deletes its oldest messages without telling anyone. With "new", JetStream refuses the publish instead; the event fails and the source retries it, e.g. the OTLP source answers with 503. Use "new" where no message may be lost to a limit, like audit logs - The retention policy controls when acknowledged messages are removed: "limits" (default) keeps all messages until stream limits like max_age, max_bytes or max_msgs are reached, "interest" removes messages when all consumers have acknowledged them, and "workqueue" removes messages when the first consumer acknowledges them. The max_age setting applies to all messages regardless of acknowledgment status - Compression saves disk space and costs CPU on the NATS server; how much it saves depends on the data. Changing it affects only data written afterwards. size counts the messages uncompressed - An event becomes one message per subject. The sink publishes them all at once and reports success once JetStream has acknowledged every one; if one fails or is not acknowledged within 10 seconds, the event fails and the source retries it, so messages already stored are stored again - Records are grouped by their resolved subject and published as one OTLP message per subject and signal type, so different signal types may share a subject. Each message has a Tram-Signal header with the signal type, which the NATS Source needs to decode it - Records keep their resource and scope. If all records of a signal type resolve to the same subject, the payload is published unchanged. Resources that appear in several entries of an event, e.g. after a batch stage, stay separate entries in the message, which OTLP allows - The register subjects configure the stream, while the CEL expressions determine the subject of each record. Records whose subject is not covered by register cannot be published - An event is rejected if a subject expression fails for one of its records, or resolves to an invalid subject: one with an empty token, whitespace or a wildcard. get_attr returns "" for a missing attribute, so 'logs.' + ...get_attr('team') gives the invalid logs. for records without it. The OTLP source answers with HTTP 422, the NATS source terminates the message instead of redelivering it - If a CEL expression fails to evaluate or returns a non-string value, the whole event is dropped with an error log and nothing of it is published - If publishing fails, the error is reported back to the source, e.g. the NATS source redelivers the message - A lost connection to NATS is retried every 2 seconds, with up to 1 second of jitter, for as long as the sink runs. While disconnected, publishing fails at once instead of being buffered, so the OTLP source answers with 503 and clients retry until NATS is back

Partitioning

A stream slows down with many subjects, and all its writes go through one NATS server. With partition, the sink spreads records over count streams by a key, e.g. the tenant, so each stream holds a part of the subjects and the streams can be led by different servers.

Field Type Required Default Description
count int Yes - Number of streams, from 1 to 1024
logs string With subject.logs - CEL expression returning the key of a log record. Has the same variables as the subject expression, without partition
metrics string With subject.metrics - CEL expression returning the key of a metric data point
traces string With subject.traces - CEL expression returning the key of a span
  • The partition of a key is its 32-bit FNV-1a hash modulo count. This never changes, so consumers can compute which stream holds a key. The NATS Source does so with its partition field
  • {partition} in stream and in every register subject is replaced by the partition number, so the sink sets up one stream per partition, e.g. audit-0 to audit-7 with the subjects audit.0.> to audit.7.>. Without partition, {partition} must not appear. All other stream settings, like replicas and max_age, apply to every stream
  • JetStream stores a message in the stream whose subjects cover it, so the subject expression has to put the partition into the subject, e.g. 'audit.' + string(partition) + '.' + .... An event with a subject that is not covered by its partition's register is rejected
  • An expression for the key that fails or does not return a string rejects the event. A missing attribute gives the key "", which is a valid key like any other
  • Each stream gets the metadata tram.partition.count and tram.partition.index. A sink refuses to start if a stream has metadata of another count or index, so streams of different partitionings are never mixed
  • Changing count moves keys to other streams. Data already stored stays in its old stream, where consumers of the new partitioning do not read it, so changing it is a migration. Choose count with room to grow

Example with partitioning by tenant:

{
  "name": "audit_sink",
  "type": "nats",
  "from": "validator",
  "spec": {
    "url": "nats://localhost:4222",
    "stream": "audit-{partition}",
    "replicas": 3,
    "partition": {
      "count": 8,
      "logs": "record.attributes.get_attr('tenant.id')"
    },
    "subject": {
      "register": ["audit.{partition}.>"],
      "logs": "'audit.' + string(partition) + '.' + record.attributes.get_attr('tenant.id')"
    }
  }
}