Stages¶

Stages process, transform, or filter data between sources and sinks.

CEL Filter Stage¶

Keeps only the records for which a Common Expression Language (CEL) expression returns true, and passes on a batch containing just those. It decides per log record, metric data point and span, so records with different attribute values that arrive in the same request, resource or scope can be separated. To accept or reject whole events instead, use the CEL Validator Stage.

Type: cel_filter

Configuration:

Field	Type	Required	Default	Description
`logs`	string	One of `logs`, `metrics`, `traces`	-	CEL expression evaluated per log record. Must return a boolean
`metrics`	string	One of `logs`, `metrics`, `traces`	-	CEL expression evaluated per metric data point. Must return a boolean
`traces`	string	One of `logs`, `metrics`, `traces`	-	CEL expression evaluated per span. Must return a boolean

Signal types without an expression are passed on unfiltered. Use "false" to remove a signal type completely.

CEL Expression Context:

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

resource and scope are empty messages when the payload has none, so their fields can be accessed without checks. Field names follow the OTLP protobuf definitions, e.g. record.severity_text, span.name, datapoint.attributes.

Example: split logs by customer into two branches:

"stages": [
  {
    "name": "customer-a",
    "type": "cel_filter",
    "from": "otlp-in",
    "spec": {
      "logs": "record.attributes.exists(a, a.key == 'customer' && a.value.string_value == 'A')"
    }
  },
  {
    "name": "other-customers",
    "type": "cel_filter",
    "from": "otlp-in",
    "spec": {
      "logs": "!record.attributes.exists(a, a.key == 'customer' && a.value.string_value == 'A')"
    }
  }
]

Additional Examples:

Keep only ERROR logs:

{ "logs": "record.severity_text == 'ERROR'" }

Keep only the data points of one metric:

{ "metrics": "metric.name == 'http.server.request.duration'" }

Keep only spans of one service:

{ "traces": "resource.attributes.exists(a, a.key == 'service.name' && a.value.string_value == 'checkout')" }

Behavior Notes: - Kept records are passed on in their original resource and scope; nothing is copied, and containers without remaining records are removed. If no record is kept, nothing is passed on - When splitting into branches, write the last branch as the negation of the others, so every record ends up in exactly one branch - An expression that fails for a record, e.g. by accessing a missing list index, rejects the whole event. The OTLP source answers with HTTP 422, the NATS source terminates the message instead of redelivering it - Place a batch stage before the filters to merge many small requests first; each branch then receives one filtered batch

CEL Validator Stage¶

Checks each incoming event as a whole with Common Expression Language (CEL) expressions and passes it on, drops it, or rejects it. An event is one delivery from the stage's parent: one OTLP request, one NATS message, or one batch. The stage never removes individual records; to keep only matching records, use the CEL Filter Stage.

Type: cel_validator

Configuration:

Field	Type	Required	Default	Description
`logs`	string	No	`"true"`	CEL expression evaluated once for all logs of an event. Must return a boolean. Empty string accepts all logs
`metrics`	string	No	`"true"`	CEL expression evaluated once for all metrics of an event. Must return a boolean. Empty string accepts all metrics
`traces`	string	No	`"true"`	CEL expression evaluated once for all traces of an event. Must return a boolean. Empty string accepts all traces
`on_mismatch`	string	Yes	-	Action when an expression returns false: `"drop"` or `"reject"`

CEL Expression Context: - Each expression receives the complete payload of its signal type: logs (LogsData), metrics (MetricsData), or traces (TracesData) - An event can contain several signal types, e.g. after a batch stage. Each type present is checked with its own expression - Use the full protobuf structure, e.g. logs.resource_logs.all(...), and write the expression as a condition that every accepted delivery has to fulfil

On Mismatch Behavior: - drop - Removes all data of the signal type whose expression returned false. Other signal types of the event are still passed on; if nothing is left, nothing is passed on - reject - Rejects the whole event with an error. The OTLP source answers with HTTP 422, the NATS source terminates the message instead of redelivering it

Example: reject log deliveries in which a resource has no service.name:

{
  "name": "require-service-name",
  "type": "cel_validator",
  "from": "source",
  "spec": {
    "logs": "logs.resource_logs.all(r, r.resource.attributes.exists(a, a.key == 'service.name'))",
    "on_mismatch": "reject"
  }
}

Additional Examples:

Drop metric deliveries that contain a metric without a name:

{
  "metrics": "metrics.resource_metrics.all(r, r.scope_metrics.all(s, s.metrics.all(m, m.name != '')))",
  "on_mismatch": "drop"
}

Reject trace deliveries that contain a span without a valid trace ID:

{
  "traces": "traces.resource_spans.all(r, r.scope_spans.all(s, s.spans.all(span, size(span.trace_id) == 16)))",
  "on_mismatch": "reject"
}

Reject log deliveries that contain a record whose tenant.id is not a lowercase UUID. matches takes an RE2 regular expression and finds it anywhere in the value, so anchor it with ^ and $. Checking a value that a NATS Sink puts into its subjects keeps out dots, whitespace and wildcards, which would make the subject invalid or change its meaning:

{
  "logs": "logs.resource_logs.all(r, r.scope_logs.all(s, s.log_records.all(record, record.attributes.exists(a, a.key == 'tenant.id' && a.value.string_value.matches('^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$')))))",
  "on_mismatch": "reject"
}

Behavior Notes: - The decision applies to the whole event. An expression like log_records.exists(record, record.severity_text == 'ERROR') passes on a delivery with one ERROR and many INFO records in full, including the INFO records - An expression that fails for an event, e.g. by accessing a missing list index, rejects the whole event regardless of on_mismatch. The OTLP source answers with HTTP 422, the NATS source terminates the message instead of redelivering it - Place validators with on_mismatch: reject before a batch stage, so one invalid delivery does not reject the deliveries batched with it

Batch Stage¶

Merges events that arrive concurrently into one event, so downstream components handle fewer, larger batches. It batches whatever it receives; use a filter stage before it to batch only some of the data.

Type: batch

Configuration:

Field	Type	Required	Default	Description
`max_records`	int	No	`8192`	Flushes once the batch holds at least this many records (log records, metric data points and spans). Events are never split, so a batch can exceed it by the size of the last event
`max_bytes`	int	No	-	Flushes a batch before the next event would take it over this many bytes of encoded OTLP data. Without it, the size is not limited
`timeout`	string	No	`"200ms"`	Flushes a batch this long after its first event arrived

Example:

{
  "name": "batch",
  "type": "batch",
  "from": "otlp-in",
  "spec": {
    "max_records": 1000,
    "timeout": "500ms"
  }
}

Behavior Notes: - Batching happens in memory only. Each caller waits until its batch has been delivered and receives the delivery result, so sources keep acknowledging or rejecting exactly what was delivered, and a slow downstream slows the sources down instead of growing the buffer - Batching only helps when events arrive concurrently, e.g. parallel OTLP requests or a NATS source with several workers. A source that sends one event at a time only adds timeout to its latency - Behind an OTLP source, timeout has to stay below the source's delivery_timeout (8s by default), together with the other batch and delay stages on the way to a sink; otherwise the stream is rejected, see Hold Times - If a caller's context is cancelled (e.g. the OTLP client disconnects or the OTLP source's delivery_timeout expires) before the batch flushes, its event is withdrawn and not delivered. Once the flush has started, the event is delivered and the caller waits for the result - A delivery error, including a rejection by a later stage, applies to every event in the batch. Place cel_validator stages with on_mismatch: reject before the batch, so one invalid event does not reject the others - max_bytes keeps batches below a limit downstream, e.g. the NATS server's max_payload (1 MB by default) or the request size a destination accepts. It counts the protobuf encoding of the data, which is what the NATS and OTLP sinks send uncompressed. An event that is larger by itself is sent alone, since events are never split, so the limit downstream can still be exceeded by single large requests. Measuring costs a pass over every event, so it is only done with max_bytes set - Events are merged by concatenating their resource entries. Resources and scopes that appear in several events are kept as separate entries, which OTLP allows - On shutdown, the open batch is delivered before downstream stages and sinks stop

Delay Stage¶

Holds every event back for a fixed duration before passing it on. The caller waits during that time, so a source that sends one event at a time is slowed down accordingly; concurrent events are delayed in parallel. Useful for testing and for simulating slow downstream systems.

Type: delay

Configuration:

Field	Type	Required	Default	Description
`delay`	string	Yes	-	Duration to hold each event back (e.g., `"100ms"`, `"1s"`, `"5m"`)

Example:

{
  "name": "delay-stage",
  "type": "delay",
  "from": "source",
  "spec": {
    "delay": "1s"
  }
}

Behavior Notes: - Behind an OTLP source, the delay counts toward the source's delivery_timeout (8s by default). A stream in which the delay and batch stages on the way to a sink add up to the delivery_timeout or more is rejected, see Hold Times. Raise the delivery_timeout for longer delays

No-op Stage¶

A pass-through stage that performs no processing. Useful for testing, debugging, or as a placeholder in pipeline designs.

Type: noop

Configuration: None required.

Example:

{
  "name": "passthrough",
  "type": "noop",
  "from": "source"
}