Pipelines

Pipeline Structure

Streams follow a tree structure with these rules:

1. Each pipeline has exactly one source (root)
2. Stages can be connected to the source or other stages
3. Sinks must be connected to either the source or a stage
4. All leaves must be sinks (no dangling stages)

Data Flow

Source → [Stage] → Sink

Arrows represent the direction of data flow, from source to sinks.

Available Components

Type	Components
Sources	generator, nats, otlp
Stages	batch, cel_filter, cel_validator, delay, noop
Sinks	debug, nats, otlp

For detailed documentation on each component and its configuration, see COMPONENTS.md.

Fan-out Pattern

Route data from a single source to multiple sinks. Every sink receives every event:

curl -X PUT http://localhost:8080/v1/streams/fanout-example \
  -H "Content-Type: application/json" \
  --data-binary @- <<'EOF'
{
  "stream": {
    "name": "fanout-example",
    "source": {
      "name": "otlp-in",
      "type": "otlp",
      "spec": { "http_port": 4318 }
    },
    "sinks": [
      {
        "name": "console",
        "type": "debug",
        "from": "otlp-in"
      },
      {
        "name": "archive",
        "type": "nats",
        "from": "otlp-in",
        "spec": {
          "url": "nats://localhost:4222",
          "stream": "telemetry",
          "subject": {
            "register": ["logs.>"],
            "logs": "'logs.' + logs.resource_logs[0].resource.attributes.get_attr('service.name')"
          }
        }
      }
    ]
  }
}
EOF

The nats sink needs a NATS server with JetStream enabled, e.g. docker compose -f setup/nats/docker-compose.yaml up -d.

Chained Stages

Process data through multiple stages before reaching sinks:

curl -X PUT http://localhost:8080/v1/streams/chained-pipeline \
  -H "Content-Type: application/json" \
  -d '{
    "stream": {
      "name": "chained-pipeline",
      "source": {
        "name": "source",
        "type": "generator",
        "spec": { "interval": "1s" }
      },
      "stages": [
        {
          "name": "stage1",
          "type": "noop",
          "from": "source"
        },
        {
          "name": "stage2",
          "type": "noop",
          "from": "stage1"
        }
      ],
      "sinks": [
        {
          "name": "output",
          "type": "debug",
          "from": "stage2"
        }
      ]
    }
  }'

Example Setup