POST /api/v1/scheduled_pipelines
scheduled_pipelines for both modes. The call requires pipelines.create.
Body
| Field | Type | Required | Notes |
|---|---|---|---|
name | string | Yes | Pipeline name. |
source_stream | string | Yes | The first stream that the pipeline reads. |
target_stream | string | Yes | The first stream that the pipeline writes. In a scheduled pipeline, a stream that the pipeline reads answers 409. |
function_steps | object | Yes | The steps, the mode, and the settings of a scheduled pipeline. See below. |
cron | string | Yes | every:30s, every:5m, or every:1h for a scheduled pipeline. An empty string for a realtime pipeline. |
lookback_secs | integer | No | Seconds that the first scheduled run of a source reads back. Default 300. A scheduled pipeline needs a positive value. |
enabled | boolean | No | Default true. |
function_steps fields:
| Field | Notes |
|---|---|
mode | scheduled (default) or realtime. |
signal_type | logs, metrics, or traces. A realtime pipeline supports logs only. |
steps | Ordered steps. A VRL step has transform_name, script, and kind set to vrl. A routing step has kind set to builtin, operation set to route, and a routing object. |
sources | The streams to read. source_stream is read first, and a scheduled pipeline also reads every other stream in the list, up to 8 in all. A realtime pipeline accepts only [source_stream]. |
sinks | The streams to write. target_stream comes first, and every other stream in the list takes every event, up to 8 in all. A realtime pipeline accepts only [target_stream]. |
sink_connectors | Scheduled only. Ids of the connectors that receive every event. |
delay_secs | Scheduled only. Seconds that the latest stretch of time is left for late events before a run reads it. 0 to 3600, default 60. |
retry | Scheduled only. run_retries, the further attempts at a window whose read or write failed, and delivery_retries, the further attempts at a connector delivery. Each 0 to 10, default 3. |
dead_letter_stream | Scheduled only. The logs stream for events that could not be processed, written, or delivered. It cannot be a stream that the pipeline reads or writes. |
dedupe_field | Scheduled only. Within one run, an event whose value of this field was already seen is dropped. |
retain_source | Realtime only. Keeps the original event in the source stream. |
curl -X POST http://localhost:5080/api/v1/scheduled_pipelines \
-H "authorization: Bearer $MS_JWT" \
-H 'content-type: application/json' \
-d '{
"name": "normalize-app",
"source_stream": "app_raw",
"target_stream": "app",
"function_steps": {
"mode": "scheduled",
"signal_type": "logs",
"sources": ["app_raw", "nginx_raw"],
"sinks": ["app", "app_archive"],
"delay_secs": 60,
"retry": { "run_retries": 3, "delivery_retries": 3 },
"dead_letter_stream": "pipeline_dead_letters",
"steps": [
{ "transform_name": "normalize-level", "kind": "vrl",
"script": ".level = downcase!(string!(.level))" }
]
},
"cron": "every:5m",
"lookback_secs": 300,
"enabled": true
}'
{
"id": "3KE6O4NTGAXQ3LAnXse8IlGpS7o",
"name": "normalize-app",
"source_stream": "app_raw",
"target_stream": "app",
"function_steps": { "mode": "scheduled", "signal_type": "logs", "sources": ["app_raw", "nginx_raw"], "steps": [ ... ] },
"cron": "every:5m",
"lookback_secs": 300,
"enabled": true,
"last_run_at_micros": null,
"read_until_micros": {},
"retry": null,
"created_at_micros": 1790000000000000,
"updated_at_micros": 1790000000000000
}
| Status | Cause |
|---|---|
400 | A schedule outside every:Ns/m/h, a non-positive lookback, a realtime pipeline for a signal other than logs, a VRL step that does not compile, an invalid stream name, more than 8 streams to read or to write, a routing step whose fixed stream or fallback stream is a source, a dead letter stream that the pipeline reads or writes, or delay_secs or a retry count out of range. |
409 | A scheduled pipeline that would write a stream that it reads, or a realtime pipeline that is already enabled for the source stream. |