POST /api/v1/scheduled_pipelines
scheduled_pipelines 这个名称。该调用需要 pipelines.create。
请求体
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
name | string | 是 | 流水线名称。 |
source_stream | string | 是 | 流水线读取的第一个数据流。 |
target_stream | string | 是 | 流水线写入的第一个数据流。定时流水线中，它是流水线读取的数据流时，调用返回 409。 |
function_steps | object | 是 | 步骤、模式以及定时流水线的设置，见下文。 |
cron | string | 是 | 定时流水线使用 every:30s、every:5m 或 every:1h；实时流水线使用空字符串。 |
lookback_secs | integer | 否 | 每个来源第一次定时运行往回读的秒数，默认 300。定时流水线需要正数。 |
enabled | boolean | 否 | 默认 true。 |
function_steps 的字段：
| 字段 | 说明 |
|---|---|
mode | scheduled（默认）或 realtime。 |
signal_type | logs、metrics 或 traces。实时流水线只支持 logs。 |
steps | 有序步骤。VRL 步骤包含 transform_name、script，并把 kind 设为 vrl。路由步骤把 kind 设为 builtin、operation 设为 route，并带有 routing 对象。 |
sources | 读取的数据流。source_stream 最先读取，定时流水线还会读取列表中的其他数据流，合计最多 8 个。实时流水线只接受 [source_stream]。 |
sinks | 写入的数据流。target_stream 排在第一位，列表中的其他数据流都收到全部事件，合计最多 8 个。实时流水线只接受 [target_stream]。 |
sink_connectors | 仅定时流水线使用，接收全部事件的连接器 id。 |
delay_secs | 仅定时流水线使用，最新一段时间留给迟到事件的秒数，过后才由运行读取。0 到 3600，默认 60。 |
retry | 仅定时流水线使用。run_retries 是读取或写入失败的窗口还会再尝试的次数，delivery_retries 是连接器投递还会再尝试的次数，各为 0 到 10，默认 3。 |
dead_letter_stream | 仅定时流水线使用，保存无法处理、写入或投递的事件的日志数据流，不能是流水线读取或写入的数据流。 |
dedupe_field | 仅定时流水线使用。同一次运行中，该字段的值已经出现过的事件会被丢弃。 |
retain_source | 仅实时流水线使用，把原始事件保留在来源数据流中。 |
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
}
| 状态码 | 原因 |
|---|---|
400 | 计划不符合 every:Ns/m/h、回看窗口不是正数、为日志之外的信号创建了实时流水线、VRL 步骤无法编译、流名不合法、读取或写入的数据流超过 8 个、路由步骤的固定数据流或兜底数据流是来源、死信数据流是流水线读取或写入的数据流，或 delay_secs、重试次数超出范围。 |
409 | 定时流水线会写入它读取的数据流，或该来源数据流已经启用了一条实时流水线。 |