Ingest pipelines

Embed
Preview of the card
<a href="https://forumulate.com/p/watchtower/docs/pipelines"><img src="https://forumulate.com/api/public/orgs/watchtower/spaces/docs/pages/170a11f5-d275-460e-a0a3-056f743c7741/social-card.svg?v=3e7ae33eec8b" alt="Ingest pipelines" width="600" height="315"></a>

A pipeline is an ordered list of stages the worker applies to every record of one signal of a project, between decode and storage. Default redaction always runs first and again at the end; it cannot be disabled. A project has one pipeline per signal.

Versions, activation and rollback

  • POST /v1/projects/{p}/pipelines {signal, name, description} creates the pipeline.

  • POST .../{id}/versions {config, note} adds version n+1. It is validated and not activated.

  • POST .../{id}/activate {version} makes a version the active one. Activating an older version is a rollback. The audit event pipeline.activate records from and to; the version keeps the previously active version as parent_version.

  • GET .../{id}/versions[/{n}] lists (newest first) and reads versions.

  • GET .../{id}/diff?from=&to= returns {stages: [{index, change, before, after}]} where change is added, removed, changed or moved.

The worker caches the active version for 30 s, so a change takes effect within about half a minute. Reads need telemetry read access; every write needs the pipeline-management permission. Pipelines of another project answer 404.

Validation

A config is {"version": 1, "stages": [...]}. A bad config answers 422; the detail is a JSON array of {path, message} with JSON pointers (/stages/3/patterns/0). Rules:

  • At most 50 stages and 50 redact patterns; unknown fields are rejected.

  • Regex patterns must compile and be at most 1 KiB. The engine is linear-time (no backtracking), so every pattern that compiles is safe to run on the ingest path.

  • Expressions (expr, quarantine.when, route.when) must parse within the expression limits.

  • sample.rate is in (0, 1]; enrich cannot set wt.* attributes.

  • A quarantine or route whose when is identical to an earlier quarantine is rejected as unreachable: the earlier stage already removed every record it could match.

Stages added in Phase 6

Stage

Config

Effect

rate_limit

key: project or attr:<name>, per_second, burst

Token bucket per key; records over the limit are dropped (outcome rate_limited). The limit applies per worker replica, so the effective limit grows with the replica count: treat it as approximate. At most 10,000 keys are tracked; further keys share one bucket.

quarantine

when (expression), reason

Matching records leave the pipeline and are produced to the signal's DLQ topic (wt.otlp.<signal>.dlq) as JSON with the tenant headers and wt-reason=quarantine:<reason>. They are never retried automatically. Delivery is at-least-once: a storage retry can quarantine a record twice.

route

when, retention_days, export_destination

Matching records are tagged with the platform attributes wt.retention_days and wt.export_destination.

Dry-run

POST .../{id}/dry-run {version | config, records} runs a stored version or an unsaved config over at most 100 log records (each at most 64 KiB) without writing anything. Records are {body, service, severity_text, severity_number, timestamp_ns, trace_id, attrs, resource}. The response is {outputs, dropped: [{index, stage}], stage_stats}. Dry-run supports logs pipelines.

Stage stats

Each worker keeps in-memory counters per project, pipeline version, stage, outcome and minute, and writes them to pipeline_stats_1m every 60 s (best effort, at most 10,000 keys between flushes). Outcomes are kept, modified, dropped, rate_limited, quarantined, aborted. Bytes in/out (estimated from attributes, resource and body) and the time spent in the stage are stored on one extra row per stage with outcome io and n = 0.

GET .../{id}/stats?from=&to=&stage= returns per-minute points (window of at most 7 days, default the last 24 hours).

Updated