Manage pipelines

curl -X POST http://localhost:5080/api/v1/scheduled_pipelines/$PIPELINE_ID/backfill \
  -H "authorization: Bearer $MS_JWT" \
  -H 'content-type: application/json' \
  -d '{"start_micros": 1790000000000000, "end_micros": 1790086400000000}'

[
  {
    "id": "3KBHcb6imgQq7FXBsROGTIRHOyg",
    "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": 1790000000000000,
    "read_until_micros": { "app_raw": 1789999940000000, "nginx_raw": 1789999940000000 },
    "retry": null,
    "created_at_micros": 1789900000000000,
    "updated_at_micros": 1789900000000000,
    "last_run_state": "succeeded",
    "last_run_started_at_micros": 1790000000000000,
    "last_run_finished_at_micros": 1790000000250000,
    "last_run_scanned_rows": 1200,
    "last_run_error": null,
    "runs_24h": 288,
    "succeeded_runs_24h": 288,
    "failed_runs_24h": 0
  }
]

curl -X POST http://localhost:5080/api/v1/scheduled_pipelines/$PIPELINE_ID/backfill \
  -H "authorization: Bearer $MS_JWT" \
  -H 'content-type: application/json' \
  -d '{"start_micros": 1790000000000000, "end_micros": 1790086400000000}'

[
  {
    "id": "3KBHcb6imgQq7FXBsROGTIRHOyg",
    "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": 1790000000000000,
    "read_until_micros": { "app_raw": 1789999940000000, "nginx_raw": 1789999940000000 },
    "retry": null,
    "created_at_micros": 1789900000000000,
    "updated_at_micros": 1789900000000000,
    "last_run_state": "succeeded",
    "last_run_started_at_micros": 1790000000000000,
    "last_run_finished_at_micros": 1790000000250000,
    "last_run_scanned_rows": 1200,
    "last_run_error": null,
    "runs_24h": 288,
    "succeeded_runs_24h": 288,
    "failed_runs_24h": 0
  }
]

Operation	Endpoint	Permission
List	`GET /api/v1/scheduled_pipelines`	`pipelines.read`
Create	`POST /api/v1/scheduled_pipelines`	`pipelines.create`
Get	`GET /api/v1/scheduled_pipelines/{id}`	`pipelines.read`
Update	`PUT /api/v1/scheduled_pipelines/{id}`	`pipelines.edit`, and `pipelines.pause` when `enabled` changes
Delete	`DELETE /api/v1/scheduled_pipelines/{id}`	`pipelines.delete`
Run history	`GET /api/v1/scheduled_pipelines/{id}/runs`	`pipelines.read`
Backfill	`POST /api/v1/scheduled_pipelines/{id}/backfill`	`pipelines.run`

List

The list returns every pipeline with the state of its latest run and a 24-hour summary:

[
  {
    "id": "3KBHcb6imgQq7FXBsROGTIRHOyg",
    "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": 1790000000000000,
    "read_until_micros": { "app_raw": 1789999940000000, "nginx_raw": 1789999940000000 },
    "retry": null,
    "created_at_micros": 1789900000000000,
    "updated_at_micros": 1789900000000000,
    "last_run_state": "succeeded",
    "last_run_started_at_micros": 1790000000000000,
    "last_run_finished_at_micros": 1790000000250000,
    "last_run_scanned_rows": 1200,
    "last_run_error": null,
    "runs_24h": 288,
    "succeeded_runs_24h": 288,
    "failed_runs_24h": 0
  }
]

Each item carries the fields of Get, the latest run as last_run_state, last_run_started_at_micros, last_run_finished_at_micros, last_run_scanned_rows (the events that it read), and last_run_error, and the counts of runs over the past 24 hours.

Get

GET /api/v1/scheduled_pipelines/{id}

Returns the pipeline with the read position of each source and the planned retry:

{
  "id": "3KBHcb6imgQq7FXBsROGTIRHOyg",
  "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": 1790000000000000,
  "read_until_micros": { "app_raw": 1789999940000000, "nginx_raw": 1789999640000000 },
  "retry": {
    "source": "nginx_raw",
    "attempt": 2,
    "next_at_micros": 1790000030000000,
    "error": "read: <reason>"
  },
  "created_at_micros": 1789900000000000,
  "updated_at_micros": 1789900000000000
}

Field	Notes
`read_until_micros`	The read position of each source, in microseconds. The events of the source before it have been handled. A source that has not been read yet has no entry.
`retry`	The soonest planned retry of a window that failed, or `null`. `source` names the source, `attempt` is the attempt that the retry will be, `next_at_micros` is when it is due, and `error` is what the last attempt failed on.

See Schedule and read positions and Retries and dead letters.

Update and delete

PUT takes the same body as Create a pipeline and replaces the configuration. Changing only enabled needs pipelines.pause, and any other change needs pipelines.edit. Changing enabled from false to true resets the read positions, so every source reads back lookback_secs again. DELETE answers {"deleted": true}.

Run history

GET /api/v1/scheduled_pipelines/{id}/runs?limit=50&before_micros=<cursor>

Returns the runs of a scheduled pipeline, newest first. A realtime pipeline records no runs.

[
  {
    "id": "3KE7Yq0d3jvXm5cJ8r2bWQzN1aP",
    "pipeline_id": "3KBHcb6imgQq7FXBsROGTIRHOyg",
    "state": "succeeded",
    "started_at_micros": 1790000000000000,
    "finished_at_micros": 1790000000250000,
    "scanned_rows": 1200,
    "written_rows": 1150,
    "dropped_rows": 48,
    "failed_rows": 2,
    "attempt": 1,
    "error": null,
    "warning": "stream `app` rejected 2 event(s): <reason>"
  }
]

Field	Notes
`state`	`running`, `succeeded`, `failed`, or `cancelled`.
`started_at_micros`, `finished_at_micros`	When the run started and ended. `finished_at_micros` is `null` while the run is running.
`scanned_rows`	The events that the run read from its sources.
`written_rows`	The events that every destination took.
`dropped_rows`	The events that a step dropped with `abort`, or that dedupe removed.
`failed_rows`	The events that a step failed on, that were routed into a source, or that a destination did not take.
`attempt`	`1` for a first attempt. Higher when the run retried a window that failed.
`error`	Why the run failed: a window that failed, with the planned retry, or a target stream that took none of its events while the pipeline has no dead letter stream.
`warning`	What went wrong without failing the run, such as events that failed, rejected or undelivered events, or a window skipped after its retries.

scanned_rows equals the sum of written_rows, dropped_rows, and failed_rows.

Backfill

POST /api/v1/scheduled_pipelines/{id}/backfill

Field	Type	Notes
`start_micros`	integer	Start of the window, in microseconds.
`end_micros`	integer	End of the window. It must be greater than `start_micros`, and the window must not exceed 31 days.

curl -X POST http://localhost:5080/api/v1/scheduled_pipelines/$PIPELINE_ID/backfill \
  -H "authorization: Bearer $MS_JWT" \
  -H 'content-type: application/json' \
  -d '{"start_micros": 1790000000000000, "end_micros": 1790086400000000}'

{
  "jobs": [
    {
      "source": "app_raw",
      "job_id": "3KE6epTrWXts1oa2ePEKIHKxG19",
      "monitor": "/api/v1/query/jobs/3KE6epTrWXts1oa2ePEKIHKxG19"
    },
    {
      "source": "nginx_raw",
      "job_id": "3KE6eqAbXq0n2yLr7VvTq9cUo4D",
      "monitor": "/api/v1/query/jobs/3KE6eqAbXq0n2yLr7VvTq9cUo4D"
    }
  ]
}

The backfill submits one search job per source. Each job reads its source over the window, applies the steps, and writes, delivers, and keeps dead letters the way a scheduled run does. The read positions of the pipeline do not move. Follow each job with its monitor path. See Query. A realtime pipeline answers 400 (realtime pipelines do not support backfill).

Last modified on October 17, 2026

Overview

Authentication

Intake

Query

APM

RUM

Streams

Alerting

Correlation & Web

Platform & Administration

List

Get

Update and delete

Run history

Backfill

​List

​Get

​Update and delete

​Run history

​Backfill

List

Get

Update and delete

Run history

Backfill