Sources

Sources are the entry points of a pipeline that produce or receive data events.

Generator Source

Emits an event at a regular interval. The events are empty: they contain no logs, metrics or traces. Useful for testing how a pipeline is wired; stages that count or inspect records, like batch or cel_filter, ignore them.

Type: generator

Configuration:

Field	Type	Required	Default	Description
interval	string	Yes	-	Time interval between events (e.g., "5s", "1m")

Example:

{
  "name": "event-generator",
  "type": "generator",
  "spec": {
    "interval": "5s"
  }
}

OTLP Source

Receives OpenTelemetry Protocol data over HTTP (OTLP/HTTP). OTLP over gRPC is not supported.

Type: otlp

Configuration:

Field	Type	Required	Default	Description
http_port	int	Yes	-	Port to listen on, on all network interfaces
delivery_timeout	string	No	"8s"	Maximum time to pass a request on to the stages and sinks, as a Go duration. A request that takes longer is cancelled and answered with 503, which clients retry
retry_after	string	No	"5s"	How long clients should wait before retrying a 503, as a Go duration. Sent as the Retry-After header in whole seconds, rounded up

Example:

{
  "name": "otlp-receiver",
  "type": "otlp",
  "spec": {
    "http_port": 4318
  }
}

Behavior Notes: - Accepts POST requests on /v1/logs, /v1/metrics and /v1/traces with a protobuf (application/x-protobuf) or JSON (application/json) body. Requests without a content type are read as protobuf - Bodies may be compressed with Content-Encoding: gzip - Each request becomes one event and is passed on as a whole - A request in which a key appears twice in the same key-value list is refused with 422: the attributes of a resource, scope, log record, span, span event, span link, data point or exemplar, a metric's metadata, and key-value lists nested in values or log bodies. OTLP requires unique keys; with a key twice, a stage could check one value while a sink routes by the other. The same key in different lists, e.g. on the resource and on a record, is allowed - Responses: 200 with an empty response message in the request's encoding once all sinks have accepted the data, 422 for a body that cannot be read or parsed, contains a duplicate key, or holds data rejected by a stage or sink (e.g. cel_validator with on_mismatch: reject), 413 for bodies larger than 20 MiB (also after decompression), 415 for other content types or encodings, 503 with a Retry-After header if passing the data on failed or took longer than delivery_timeout. Only the 503 is retried by OTLP clients: it is the only answer the request itself did not cause - A rejection is answered with 422 wherever it comes from, including a sink whose destination refuses the data permanently, e.g. an OTLP Sink whose destination answers 401, 403 or 404 because of its own configuration. The client then drops data it did not cause to be refused - The default delivery_timeout is below the 10s that OpenTelemetry SDK exporters wait by default, so they get the 503 before they give up. batch and delay stages have to leave time for it, see Hold Times. Raise it for pipelines that take longer, e.g. with a delay stage or slow sinks, and for clients that wait longer, like the OpenTelemetry Collector (30s by default). A batch that is already being sent when the timeout expires is still awaited, since it holds data of other requests too - A request may take up to 1 minute to send, which leaves time for large bodies on slow connections - The port is opened when the stream is created, so a port that is already in use fails the stream creation

NATS Source

Consumes events from a NATS JetStream stream that were published by Tram's NATS Sink.

Type: nats

Configuration:

Field	Type	Required	Default	Description
url	string	Yes	-	NATS server URL (e.g., "nats://localhost:4222")
stream	string	Yes	-	JetStream stream to consume from. It must exist when the Tram stream is created; otherwise the source logs an error and stays idle until the Tram stream is updated
subject	string	Yes	-	Subject filter of the consumer; wildcards like logs.* are allowed
durable	string	No	-	Name of a durable consumer. Without it, a temporary consumer is created
workers	int	No	1	Number of messages processed concurrently (1-10000)
partition	object	No	-	Reads the stream of a key from streams partitioned by the NATS sink: count (1-1024) as in the sink, and the required key

Example:

{
  "name": "nats-consumer",
  "type": "nats",
  "spec": {
    "url": "nats://localhost:4222",
    "stream": "otlp-data",
    "subject": "logs.*"
  }
}

Example with durable consumer and multiple workers:

{
  "name": "nats-consumer",
  "type": "nats",
  "spec": {
    "url": "nats://localhost:4222",
    "stream": "otlp-data",
    "subject": "logs.*",
    "durable": "tram-consumer",
    "workers": 5
  }
}

Behavior Notes: - Only messages with a Tram-Signal header, which Tram's NATS sink sets, can be decoded. Other messages are terminated and not redelivered - Each message becomes one event - A message is acknowledged once all sinks have accepted it. If passing it on fails, it is redelivered after 5 seconds; data rejected by a stage is terminated instead - Delivery is at-least-once: when one branch of a stream fails, the redelivered message reaches all branches again - The consumer starts at the beginning of the stream. A durable consumer continues where it stopped; a temporary consumer starts from the beginning again whenever the source is created, i.e. when the Tram stream is created or its source is changed. Updates that only change stages or sinks keep the consumer - Up to workers messages are processed at the same time, each on its own. The source fetches only as many messages as workers are free, so no fetched message waits for a worker while its acknowledgement deadline runs - Messages are not processed in the order of the stream, unless workers is 1 - A batch stage after the source holds up to workers messages, since every message waits for its batch to be delivered before it is acknowledged. To let batches fill up to max_records before their timeout, set workers to at least max_records divided by the records per message - The source sets the consumer's MaxAckPending, the unacknowledged messages JetStream hands out, to twice workers and at least 1,000, the JetStream default. The limit applies across all Tram replicas that share a durable consumer - Stopping or updating the source interrupts waiting for new messages right away; it only waits for messages that are being processed, which are still delivered and acknowledged - A lost connection to NATS is retried every 2 seconds, with up to 1 second of jitter, for as long as the source runs, so the source continues after any outage. Acknowledgements that fail while disconnected are not repeated; those messages are redelivered after their ack wait - With partition, the source computes the partition of key like the NATS sink and replaces {partition} in stream and subject with it, so a source per key reads only the stream that holds the key. stream must contain {partition}; without partition, neither field may contain it - count must match the sink's. The source refuses a stream whose metadata names another count or partition and stays idle like with a missing stream. With a wrong count, the key's stream may also not exist at all

Example with partitioning by tenant:

{
  "name": "tenant-123",
  "type": "nats",
  "spec": {
    "url": "nats://localhost:4222",
    "stream": "audit-{partition}",
    "subject": "audit.{partition}.tenant-123",
    "durable": "tenant-123",
    "partition": {
      "count": 8,
      "key": "tenant-123"
    }
  }
}