mirror of
https://github.com/prdlk/dotfiles.git
synced 2026-11-06 12:37:37 +00:00
config(git,claude,ssh): templated gitconfig; claude settings/commands/hooks/skills; ssh config
gitconfig: name/email from chezmoi data, gh credential helper by PATH lookup (portable to darwin). claude: no credentials, history, projects, sessions, or plugin caches. ssh config as private_ (0600). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
3c691e8dc5
commit
20891f3910
476 files changed
+59086
-7
No files matched your search
@@ -0,0 +1,90 @@
|
||||
# Cloudflare Pipelines
|
||||
|
||||
Streaming ingest: receive events over HTTP/Workers/Logpush, transform with SQL, write to R2 as Iceberg tables or Parquet/JSON files.
|
||||
|
||||
## Documentation
|
||||
|
||||
This reference is a fast-start with verified code and gotchas. For limits, settings, full SQL syntax, and pricing, **retrieve the live docs** — use the `cloudflare-docs` MCP/search tool if available, otherwise `webfetch` the URL. Docs are source of truth over this file.
|
||||
|
||||
| Topic | URL |
|
||||
|-------|-----|
|
||||
| Overview / getting started | `https://developers.cloudflare.com/pipelines/getting-started/` |
|
||||
| Streams (write, manage, Logpush) | `https://developers.cloudflare.com/pipelines/streams/` |
|
||||
| Sinks | `https://developers.cloudflare.com/pipelines/sinks/` |
|
||||
| Pipelines & SQL transforms | `https://developers.cloudflare.com/pipelines/pipelines/` |
|
||||
| SQL reference (statements, types) | `https://developers.cloudflare.com/pipelines/sql-reference/` |
|
||||
| Wrangler commands | `https://developers.cloudflare.com/pipelines/reference/wrangler-commands/` |
|
||||
| Terraform | `https://developers.cloudflare.com/pipelines/reference/terraform/` |
|
||||
| Limits | `https://developers.cloudflare.com/pipelines/platform/limits/` |
|
||||
| Pricing | `https://developers.cloudflare.com/pipelines/platform/pricing/` |
|
||||
| Metrics (GraphQL) | `https://developers.cloudflare.com/pipelines/observability/metrics/` |
|
||||
|
||||
## Three Components
|
||||
|
||||
```
|
||||
Sources → Stream → Pipeline (SQL) → Sink → R2
|
||||
↑ ↓ ↓
|
||||
HTTP / Workers / Transform Iceberg (Data Catalog)
|
||||
Logpush (row-level) or Parquet/JSON files
|
||||
```
|
||||
|
||||
| Component | Purpose |
|
||||
|-----------|---------|
|
||||
| **Stream** | Receives events (HTTP endpoint, Worker binding, or Logpush). Structured (schema-validated) or unstructured. |
|
||||
| **Pipeline** | SQL connecting a stream to a sink. Row-level transforms only — no GROUP BY/aggregation. |
|
||||
| **Sink** | Writes to R2 — Iceberg via Data Catalog, or raw Parquet/JSON. |
|
||||
|
||||
**Status:** Open beta (Workers Paid for production). Pricing announced; verify billing status in docs.
|
||||
|
||||
## Quick Start
|
||||
|
||||
```bash
|
||||
# Interactive — creates stream + sink + pipeline, optionally bucket + catalog
|
||||
npx wrangler pipelines setup
|
||||
```
|
||||
|
||||
Minimal Worker producer:
|
||||
```typescript
|
||||
interface Env { MY_STREAM: Pipeline; }
|
||||
|
||||
export default {
|
||||
async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
|
||||
ctx.waitUntil(env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }]));
|
||||
return new Response("OK");
|
||||
}
|
||||
} satisfies ExportedHandler<Env>;
|
||||
```
|
||||
|
||||
## Which Sink Type?
|
||||
|
||||
```
|
||||
Need SQL queries / ACID / time-travel on the data?
|
||||
→ R2 Data Catalog (Iceberg) ✅ R2 SQL, schema evolution ❌ more setup
|
||||
|
||||
Just archival / external tools (Spark, Athena)?
|
||||
→ R2 raw files (Parquet/JSON) ✅ simple, partitioned files ❌ no built-in SQL
|
||||
```
|
||||
|
||||
## Critical Behaviors (read before building)
|
||||
|
||||
These are non-obvious and prevent most failures — see [gotchas.md](gotchas.md) for detail.
|
||||
|
||||
- **Everything is immutable after creation** — stream schema, pipeline SQL, sink config. To change, delete and recreate.
|
||||
- **Sinks create their own table** — they cannot target an existing Iceberg table.
|
||||
- **`__ingest_ts` is added automatically** (TIMESTAMP, partitioned by day). Don't define it in your schema.
|
||||
- **Data isn't queryable immediately** — first flush takes **3–7 minutes** (warm-up + table creation) even with a short roll interval.
|
||||
- **Schema validation is deferred** — invalid events are accepted then silently dropped. Monitor via GraphQL error metrics.
|
||||
- **Binding field renamed `pipeline` → `stream`** (June 2026); old field still accepted.
|
||||
|
||||
## Reading Order
|
||||
|
||||
1. [configuration.md](configuration.md) — schema, streams, sinks, pipelines (CLI + REST + Terraform), bindings
|
||||
2. [api.md](api.md) — `send()`, HTTP ingest, REST API, pipeline SQL, lifecycle states
|
||||
3. [patterns.md](patterns.md) — fire-and-forget, validation, Logpush, observability, end-to-end
|
||||
4. [gotchas.md](gotchas.md) — silent drops, immutability, REST≠CLI field names
|
||||
|
||||
## See Also
|
||||
|
||||
- [r2-data-catalog](../r2-data-catalog/) — Iceberg sink destination
|
||||
- [r2-sql](../r2-sql/) — query the ingested data
|
||||
- [r2](../r2/) · [queues](../queues/) · [workers](../workers/)
|
||||
@@ -0,0 +1,124 @@
|
||||
# Pipelines API Reference
|
||||
|
||||
Code templates and verified behavior. For the full SQL function set and HTTP status semantics, pull `https://developers.cloudflare.com/pipelines/sql-reference/` and the streams docs.
|
||||
|
||||
## Worker Binding Interface
|
||||
|
||||
```typescript
|
||||
// from cloudflare:pipelines / @cloudflare/workers-types
|
||||
interface Pipeline<T = any> { send(records: T[]): Promise<void>; }
|
||||
|
||||
interface Env { MY_STREAM: Pipeline; }
|
||||
|
||||
export default {
|
||||
async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
|
||||
await env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }]);
|
||||
return new Response("OK");
|
||||
}
|
||||
} satisfies ExportedHandler<Env>;
|
||||
```
|
||||
|
||||
- `send()` takes an **array**, returns `Promise<void>` (no confirmation payload).
|
||||
- Throws on network errors — wrap in try/catch or use `ctx.waitUntil()` for fire-and-forget.
|
||||
- Validation errors are **not** thrown here (deferred during processing — see [gotchas.md](gotchas.md)).
|
||||
- Payload/rate limits apply — check `https://developers.cloudflare.com/pipelines/platform/limits/` before sizing batches.
|
||||
|
||||
## HTTP Ingest
|
||||
|
||||
```
|
||||
https://{stream-id}.ingest.cloudflare.com
|
||||
```
|
||||
|
||||
Get `{stream-id}` from `npx wrangler pipelines streams list`.
|
||||
|
||||
```bash
|
||||
# Batch (preferred)
|
||||
curl -X POST https://{stream-id}.ingest.cloudflare.com \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '[{"event_id":"evt-1","amount":29.99},{"event_id":"evt-2","amount":14.99}]'
|
||||
|
||||
# Single event — auto-wrapped in an array
|
||||
curl -X POST https://{stream-id}.ingest.cloudflare.com \
|
||||
-H "Content-Type: application/json" -d '{"event_id":"evt-3","amount":9.99}'
|
||||
```
|
||||
|
||||
If stream auth is enabled, add `-H "Authorization: Bearer $TOKEN"` (token needs **Workers Pipelines Send**). Standard HTTP status codes apply (400 invalid, 401 auth, 413 too large, 429 rate-limited, 5xx retry).
|
||||
|
||||
> **JSON only** — no Avro, Protobuf, or CSV input.
|
||||
|
||||
## REST Management API
|
||||
|
||||
Base: `https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/pipelines/v1`
|
||||
|
||||
```bash
|
||||
# List
|
||||
curl -s "$BASE_URL/streams" -H "Authorization: Bearer $API_TOKEN"
|
||||
curl -s "$BASE_URL/sinks" -H "Authorization: Bearer $API_TOKEN"
|
||||
curl -s "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN"
|
||||
|
||||
# Get one (pipeline GET includes status + failure_reason — useful for debugging)
|
||||
curl -s "$BASE_URL/pipelines/{pipeline-id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
|
||||
# Delete in reverse order: pipeline → sink → stream
|
||||
curl -X DELETE "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
curl -X DELETE "$BASE_URL/sinks/{id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
curl -X DELETE "$BASE_URL/streams/{id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
```
|
||||
|
||||
> `wrangler pipelines delete` defaults to "no" non-interactively — use the REST API for automated cleanup. Deleting a stream removes buffered events and dependent pipelines.
|
||||
|
||||
### Pipeline Lifecycle States
|
||||
|
||||
| Status | Meaning |
|
||||
|--------|---------|
|
||||
| `running` | Active, processing events |
|
||||
| `initializing` | Starting up (minutes after creation or recovery) |
|
||||
| `failed` | Stopped on error — check `failure_reason` (expired token, deleted bucket, disabled catalog) |
|
||||
|
||||
> A `GET` on a sink shows `schema.fields: []` — expected. The sink inherits schema from the stream via the pipeline SQL.
|
||||
|
||||
## Pipeline SQL (Transforms)
|
||||
|
||||
Row-level only — no GROUP BY/aggregation. CTEs (`WITH`) and `UNNEST` are supported. Full function list: `https://developers.cloudflare.com/pipelines/sql-reference/`.
|
||||
|
||||
```sql
|
||||
-- Passthrough / filter / enrich
|
||||
INSERT INTO my_sink SELECT * FROM my_stream;
|
||||
INSERT INTO my_sink SELECT * FROM my_stream WHERE amount > 10;
|
||||
INSERT INTO my_sink
|
||||
SELECT event_id, UPPER(category) AS category, amount * 1.1 AS amount_with_tax
|
||||
FROM my_stream;
|
||||
|
||||
-- CTE
|
||||
WITH filtered AS (SELECT event_id, amount FROM my_stream WHERE amount > 50)
|
||||
INSERT INTO my_sink SELECT * FROM filtered;
|
||||
|
||||
-- UNNEST arrays (one per SELECT)
|
||||
SELECT UNNEST(tags) AS tag FROM my_stream;
|
||||
```
|
||||
|
||||
Supported categories: string, regex, hashing (`sha256`), JSON extraction, timestamp conversion, conditional (`CASE`), `CAST`, `COALESCE`, math/comparison operators.
|
||||
|
||||
## Verifying End-to-End Data Flow
|
||||
|
||||
```bash
|
||||
# 1. Pipeline running (not initializing/failed)?
|
||||
curl -s "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
|
||||
# 2. Table created yet? (3–7 min on first flush)
|
||||
curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/namespaces/my_ns/tables" \
|
||||
-H "Authorization: Bearer $API_TOKEN"
|
||||
|
||||
# 3. Data present? (R2 SQL)
|
||||
curl -s -X POST \
|
||||
"https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/r2-sql/query/$BUCKET" \
|
||||
-H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \
|
||||
-d '{"query": "SELECT COUNT(*) AS total FROM my_ns.my_table"}'
|
||||
```
|
||||
|
||||
> Expect **3–7 minutes** from first send to first queryable data. Subsequent flushes are much faster.
|
||||
|
||||
## See Also
|
||||
|
||||
- [configuration.md](configuration.md) — creating resources · [patterns.md](patterns.md) — producers, Logpush, observability
|
||||
- [r2-sql/api.md](../r2-sql/api.md) — querying results
|
||||
@@ -0,0 +1,155 @@
|
||||
# Pipelines Configuration
|
||||
|
||||
Templates for creating streams, sinks, and pipelines via CLI, REST, or Terraform. For the full flag/field list and allowed values, pull `https://developers.cloudflare.com/pipelines/reference/wrangler-commands/` and the streams/sinks/pipelines docs.
|
||||
|
||||
## Naming Rules
|
||||
|
||||
- **Streams, sinks, pipelines** use underscores: `my_stream`, `my_sink`, `my_pipeline`.
|
||||
- **Buckets** use hyphens: `my-bucket`.
|
||||
|
||||
## Schema (Structured Streams)
|
||||
|
||||
Schema is a JSON object with a `fields` array; each field has `name`, `type`, `required`.
|
||||
|
||||
```json
|
||||
{
|
||||
"fields": [
|
||||
{ "name": "event_id", "type": "string", "required": true },
|
||||
{ "name": "amount", "type": "float64", "required": false }
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
Field types include `string`, `bool`, `int32/64`, `float32/64`, `timestamp`, `json`, `binary`, `list`, `struct` (with nested `items`/`fields`). For the authoritative type list, see `https://developers.cloudflare.com/pipelines/sql-reference/sql-data-types/`.
|
||||
|
||||
Unstructured streams (no schema) store everything in a single `value` column.
|
||||
|
||||
> Pipelines auto-adds `__ingest_ts` (TIMESTAMP, day-partitioned). Do **not** include it in your schema.
|
||||
|
||||
## Option A: Interactive (Simplest)
|
||||
|
||||
```bash
|
||||
npx wrangler pipelines setup # creates stream + sink + pipeline, optionally bucket + catalog
|
||||
```
|
||||
|
||||
## Option B: Wrangler CLI (Explicit)
|
||||
|
||||
```bash
|
||||
# 1. Stream
|
||||
npx wrangler pipelines streams create my_stream --schema-file schema.json
|
||||
|
||||
# 2. Sink — R2 Data Catalog (Iceberg). Creates the namespace + table.
|
||||
npx wrangler pipelines sinks create my_sink \
|
||||
--type r2-data-catalog \
|
||||
--bucket my-bucket --namespace my_namespace --table my_table \
|
||||
--catalog-token $API_TOKEN \
|
||||
--compression zstd --roll-interval 300
|
||||
|
||||
# 2b. Sink — R2 raw Parquet (alternative)
|
||||
npx wrangler pipelines sinks create my_sink \
|
||||
--type r2 --bucket my-bucket --format parquet \
|
||||
--path analytics/events --partitioning "year=%Y/month=%m/day=%d" \
|
||||
--access-key-id $KEY --secret-access-key $SECRET
|
||||
|
||||
# 3. Pipeline (SQL connects stream → sink)
|
||||
npx wrangler pipelines create my_pipeline \
|
||||
--sql "INSERT INTO my_sink SELECT * FROM my_stream"
|
||||
```
|
||||
|
||||
Tuning knobs (`--compression`, `--roll-interval`, `--roll-size`, etc.) and their allowed values/defaults change — pull the wrangler-commands and sinks docs rather than hardcoding. Rule of thumb: prod `--roll-interval 300+`, dev `10` (creates many small files).
|
||||
|
||||
> **⚠️ Pipelines are immutable.** SQL, schema, and sink config can't be changed — delete and recreate.
|
||||
|
||||
## Option C: REST API (Programmatic)
|
||||
|
||||
Base: `https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/pipelines/v1`
|
||||
|
||||
```bash
|
||||
# Stream
|
||||
curl -X POST "$BASE_URL/streams" -H "Authorization: Bearer $API_TOKEN" \
|
||||
-H "Content-Type: application/json" -d '{
|
||||
"name": "my_stream",
|
||||
"http": {"enabled": true, "authentication": false},
|
||||
"schema": {"fields": [{"name": "event_id", "type": "string", "required": true}]}
|
||||
}'
|
||||
|
||||
# Sink — NOTE REST field names differ from CLI flags (see table)
|
||||
curl -X POST "$BASE_URL/sinks" -H "Authorization: Bearer $API_TOKEN" \
|
||||
-H "Content-Type: application/json" -d '{
|
||||
"name": "my_sink", "type": "r2_data_catalog",
|
||||
"config": {"bucket": "my-bucket", "namespace": "my_namespace",
|
||||
"table_name": "my_table", "token": "'$API_TOKEN'",
|
||||
"rolling_policy": {"interval_seconds": 300}},
|
||||
"format": {"type": "parquet"}
|
||||
}'
|
||||
|
||||
# Pipeline
|
||||
curl -X POST "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"name": "my_pipeline", "sql": "INSERT INTO my_sink SELECT * FROM my_stream;"}'
|
||||
```
|
||||
|
||||
**REST field names ≠ CLI flags** (common failure — not obvious from docs):
|
||||
|
||||
| REST (config body) | CLI flag | Gotcha |
|
||||
|--------------------|----------|--------|
|
||||
| `"type": "r2_data_catalog"` | `--type r2-data-catalog` | underscores vs hyphens |
|
||||
| `"table_name"` | `--table` | different key |
|
||||
| `"token"` | `--catalog-token` | different key |
|
||||
| `"format": {"type": "parquet"}` | (implied) | required in REST, omitted in CLI |
|
||||
|
||||
## Worker Binding
|
||||
|
||||
```jsonc
|
||||
// wrangler.jsonc
|
||||
{ "pipelines": [ { "stream": "<STREAM_ID>", "binding": "MY_STREAM" } ] }
|
||||
```
|
||||
|
||||
> Binding field is `"stream"` as of June 2026 (was `"pipeline"`, still accepted). Use the **stream ID** (`wrangler pipelines streams list`), not the pipeline ID. Redeploy after adding. Generate typed bindings with `npx wrangler types` → `Pipeline<Cloudflare.MyStreamRecord>` from `cloudflare:pipelines`.
|
||||
|
||||
## Terraform
|
||||
|
||||
Resources: `cloudflare_pipeline_stream`, `cloudflare_pipeline_sink`, `cloudflare_pipeline`. For current attribute schemas pull `https://developers.cloudflare.com/pipelines/reference/terraform/`.
|
||||
|
||||
```hcl
|
||||
resource "cloudflare_pipeline_stream" "my_stream" {
|
||||
account_id = var.cloudflare_account_id
|
||||
name = "my_stream"
|
||||
format = { type = "json" }
|
||||
schema = { fields = [{ name = "value", type = "json", required = true }] }
|
||||
http = { enabled = true, authentication = false, cors = {} }
|
||||
worker_binding = { enabled = false }
|
||||
}
|
||||
|
||||
resource "cloudflare_pipeline_sink" "my_sink" {
|
||||
account_id = var.cloudflare_account_id
|
||||
name = "my_sink"
|
||||
type = "r2_data_catalog"
|
||||
format = { type = "parquet" }
|
||||
schema = { fields = [] }
|
||||
config = {
|
||||
account_id = var.cloudflare_account_id
|
||||
bucket = cloudflare_r2_bucket.pipeline_bucket.name
|
||||
table_name = "my_table"
|
||||
token = var.catalog_token
|
||||
}
|
||||
}
|
||||
|
||||
resource "cloudflare_pipeline" "my_pipeline" {
|
||||
account_id = var.cloudflare_account_id
|
||||
name = "my_pipeline"
|
||||
sql = "INSERT INTO ${cloudflare_pipeline_sink.my_sink.name} SELECT * FROM ${cloudflare_pipeline_stream.my_stream.name}"
|
||||
}
|
||||
```
|
||||
|
||||
## Credentials
|
||||
|
||||
| Type | Permission |
|
||||
|------|------------|
|
||||
| Catalog token (Iceberg sink) | R2 Storage Admin R&W + R2 Data Catalog R&W |
|
||||
| R2 credentials (raw sink) | Object Read & Write |
|
||||
| HTTP ingest token | Workers Pipelines Send (only if stream auth enabled) |
|
||||
|
||||
## See Also
|
||||
|
||||
- [api.md](api.md) — sending events, REST API, lifecycle · [gotchas.md](gotchas.md) — immutability, REST≠CLI
|
||||
@@ -0,0 +1,58 @@
|
||||
# Pipelines Gotchas
|
||||
|
||||
Non-obvious failure modes (not well covered by docs). For current limits and error semantics, pull `https://developers.cloudflare.com/pipelines/platform/limits/`.
|
||||
|
||||
## Events accepted but never appear (most common)
|
||||
|
||||
HTTP 200 / `send()` resolves, but no data in the sink. Causes:
|
||||
|
||||
1. **Schema validation failure** — structured streams accept then **silently drop** invalid events during processing. Validate client-side (Zod) and monitor `pipelinesUserErrorsAdaptiveGroups`.
|
||||
2. **First-flush warm-up** — first data takes **3–7 minutes** (warm-up + namespace/table creation) even with `--roll-interval 10`. Poll ≥5 min in tests.
|
||||
3. **Roll interval not elapsed** — default 300s.
|
||||
4. **Silent sink failure** — deleted bucket or expired token. Check `recordsWritten > 0` but `filesWritten = 0`; inspect `failure_reason` via `GET /pipelines/{id}`.
|
||||
|
||||
## Everything is immutable
|
||||
|
||||
Cannot modify stream schema, pipeline SQL, or sink config — delete and recreate. Use version naming (`events_v1`) and keep SQL in version control.
|
||||
|
||||
```bash
|
||||
curl -X DELETE "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
curl -X DELETE "$BASE_URL/sinks/{id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
curl -X DELETE "$BASE_URL/streams/{id}" -H "Authorization: Bearer $API_TOKEN"
|
||||
```
|
||||
|
||||
## Worker binding undefined (`env.MY_STREAM`)
|
||||
|
||||
1. Use the **stream ID**, not pipeline ID, in `wrangler.jsonc`.
|
||||
2. Binding field is `"stream"` (June 2026); old `"pipeline"` still works.
|
||||
3. Redeploy after adding the binding.
|
||||
|
||||
## REST API field names ≠ CLI flags
|
||||
|
||||
`r2_data_catalog` vs `--type r2-data-catalog`, `table_name` vs `--table`, `token` vs `--catalog-token`, and `format` is required in REST but implied in CLI. See [configuration.md](configuration.md#option-c-rest-api-programmatic).
|
||||
|
||||
## `wrangler pipelines delete` defaults to "no"
|
||||
|
||||
Non-interactive environments answer "no" automatically — use REST `DELETE` for CI/automation.
|
||||
|
||||
## Behavioral Notes
|
||||
|
||||
- **`__ingest_ts` auto-added** (TIMESTAMP, day-partitioned). Don't put it in your schema.
|
||||
- **Sinks can't target existing tables** — the sink creates its own. Use PySpark to write to existing tables.
|
||||
- **JSON-only input** — no Avro/Protobuf/CSV.
|
||||
- **Naming:** streams/sinks/pipelines use underscores; buckets use hyphens.
|
||||
- **Metrics lag 5–10 min** after creation.
|
||||
- **Pipeline SQL is row-level only** — no GROUP BY/aggregation/window functions (do aggregation in [R2 SQL](../r2-sql/) at query time). CTEs and `UNNEST` are supported.
|
||||
|
||||
## Debug Checklist
|
||||
|
||||
- [ ] Stream exists: `wrangler pipelines streams list`
|
||||
- [ ] Pipeline `running` (not `initializing`/`failed`): `GET /pipelines/{id}`, check `failure_reason`
|
||||
- [ ] SQL matches schema; sink token valid; bucket + catalog exist
|
||||
- [ ] Worker redeployed; binding uses **stream ID** under `"stream"`
|
||||
- [ ] Waited ≥5 min (first flush)
|
||||
- [ ] Sink metrics: `filesWritten > 0`; error metrics show no drops
|
||||
|
||||
## See Also
|
||||
|
||||
- [configuration.md](configuration.md) · [api.md](api.md) · [patterns.md](patterns.md)
|
||||
@@ -0,0 +1,130 @@
|
||||
# Pipelines Patterns
|
||||
|
||||
Code-first patterns. For observability dataset/field schemas and Logpush dataset lists, pull `https://developers.cloudflare.com/pipelines/observability/metrics/` and `https://developers.cloudflare.com/pipelines/streams/logpush/`.
|
||||
|
||||
## Fire-and-Forget Producer
|
||||
|
||||
```typescript
|
||||
export default {
|
||||
async fetch(req, env, ctx) {
|
||||
const event = { event_id: crypto.randomUUID(), event_type: "page_view", timestamp: new Date().toISOString() };
|
||||
ctx.waitUntil(env.MY_STREAM.send([event])); // don't block the response
|
||||
return new Response("OK");
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
## Client-Side Validation with Zod
|
||||
|
||||
Structured streams drop invalid events silently during processing. Validate before sending for immediate feedback.
|
||||
|
||||
```typescript
|
||||
import { z } from "zod";
|
||||
|
||||
const EventSchema = z.object({
|
||||
event_id: z.string(),
|
||||
category: z.enum(["purchase", "view"]),
|
||||
amount: z.number().positive().optional(),
|
||||
});
|
||||
|
||||
const validated = EventSchema.parse(rawEvent); // throws synchronously
|
||||
await env.MY_STREAM.send([validated]);
|
||||
```
|
||||
|
||||
## Scheduled Collector Worker
|
||||
|
||||
```jsonc
|
||||
// wrangler.jsonc
|
||||
{
|
||||
"name": "collector",
|
||||
"pipelines": [{ "stream": "<STREAM_ID>", "binding": "EVENT_STREAM" }],
|
||||
"triggers": { "crons": ["*/5 * * * *"] }
|
||||
}
|
||||
```
|
||||
|
||||
```typescript
|
||||
export default {
|
||||
async scheduled(event, env, ctx) {
|
||||
const items = await (await fetch("https://api.example.com/data")).json();
|
||||
const events = items.map(i => ({
|
||||
event_id: crypto.randomUUID(),
|
||||
timestamp: new Date().toISOString(),
|
||||
category: i.type, amount: i.value,
|
||||
}));
|
||||
await env.EVENT_STREAM.send(events);
|
||||
},
|
||||
};
|
||||
```
|
||||
|
||||
## Logpush → Pipelines
|
||||
|
||||
Pipelines is a native Logpush destination — ingest Cloudflare logs, transform with SQL, store as Iceberg/Parquet. For the current supported dataset list and field names, pull the Logpush doc above.
|
||||
|
||||
```sql
|
||||
INSERT INTO http_logs_sink
|
||||
SELECT
|
||||
ClientIP,
|
||||
EdgeResponseStatus,
|
||||
to_timestamp_micros(EdgeStartTimestamp) AS event_time,
|
||||
upper(ClientRequestMethod) AS method,
|
||||
sha256(ClientIP) AS hashed_ip -- redact PII at ingest
|
||||
FROM http_logs_stream
|
||||
WHERE EdgeResponseStatus >= 400;
|
||||
```
|
||||
|
||||
Configure via Dashboard (**Logpush → Create a job → Pipelines** destination) or API.
|
||||
|
||||
## Pipelines + Queues Fan-out
|
||||
|
||||
```typescript
|
||||
await Promise.all([
|
||||
env.ANALYTICS_STREAM.send([event]), // long-term storage + SQL
|
||||
env.PROCESS_QUEUE.send(event), // immediate processing + retries
|
||||
]);
|
||||
```
|
||||
|
||||
Use Pipelines for long-term storage + SQL; Queues for immediate processing/retries/DLQ; both for fan-out.
|
||||
|
||||
## Observability (GraphQL Analytics)
|
||||
|
||||
Same R2 API token works. Endpoint: `https://api.cloudflare.com/client/v4/graphql`. Datasets cover ingestion, processing (incl. `decodeErrors`), delivery, sink writes (`filesWritten`), and user/validation errors — see the metrics doc for the full dataset/field catalog.
|
||||
|
||||
```bash
|
||||
curl -X POST "https://api.cloudflare.com/client/v4/graphql" \
|
||||
-H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \
|
||||
-d '{"query": "query { viewer { accounts(filter: {accountTag: \"'$ACCOUNT_ID'\"}) { pipelinesIngestionAdaptiveGroups(filter: {pipelineId: \"PIPELINE-UUID-WITH-DASHES\", datetime_geq: \"2026-03-01T00:00:00Z\"}, limit: 10) { sum { ingestedRecords ingestedBytes } dimensions { datetimeHour } } } } }"}'
|
||||
```
|
||||
|
||||
> **Sink/pipeline IDs need dashes for GraphQL** but wrangler may show them without: `b909fe6e544844abbd63f6dcbc81d602` → `b909fe6e-5448-44ab-bd63-f6dcbc81d602`. Metrics take 5–10 min to populate.
|
||||
|
||||
### Detecting Silent Data Loss
|
||||
|
||||
If a sink's bucket is deleted or its token expires, events are accepted but lost. Tell-tale: `recordsWritten > 0` but `filesWritten = 0`. Always verify data lands in R2 within the roll interval and R2 SQL returns expected counts.
|
||||
|
||||
## Schema Evolution (Immutable Pipelines)
|
||||
|
||||
Pipelines can't change. Version + dual-write:
|
||||
|
||||
```bash
|
||||
npx wrangler pipelines streams create events_v2 --schema-file v2.json
|
||||
```
|
||||
```typescript
|
||||
await Promise.all([env.EVENTS_V1.send([event]), env.EVENTS_V2.send([event])]);
|
||||
// query across versions with UNION ALL in R2 SQL
|
||||
```
|
||||
|
||||
## End-to-End: Streaming Analytics Dashboard
|
||||
|
||||
```
|
||||
External APIs → Collector Worker (cron) → Pipeline → R2 (Iceberg) → Dashboard Worker → R2 SQL
|
||||
```
|
||||
|
||||
1. Create bucket + enable catalog ([r2-data-catalog](../r2-data-catalog/configuration.md))
|
||||
2. Create stream + sink + pipeline (here)
|
||||
3. Collector Worker with cron + stream binding (above)
|
||||
4. Dashboard Worker querying R2 SQL ([r2-sql/patterns.md](../r2-sql/patterns.md))
|
||||
5. Enable automatic compaction
|
||||
|
||||
## See Also
|
||||
|
||||
- [configuration.md](configuration.md) · [api.md](api.md) · [gotchas.md](gotchas.md) · [r2-sql](../r2-sql/)
|
||||
Reference in new issue
Block a user