# Tune Pipeline > Use when a GlassFlow pipeline is experiencing backpressure, sustained sink lag, or throughput pain. Reads GlassFlow's emitted Prometheus metrics to confirm the bottleneck (backpressure gauges, stream depth ratio, source vs sink rate, sink batch size distribution), proposes targeted changes to replicas, batch sizes, stream limits, or dedup storage, applies them via the appropriate API endpoint, and verifies improvement by re-reading the same metrics. Triggers on phrases like "pipeline has backpressure", "tune pipeline X", "throughput is too low", "size resources for pipeline Y". - Skill: `glassflow/tune-pipeline` (Agent Skill) - Install (CLI): `npx skillmds@latest add glassflow/tune-pipeline` - Raw SKILL.md: https://api.skillmd.com/api/skills/glassflow/tune-pipeline/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - Author: glassflow (https://skillmd.com/u/glassflow) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/glassflow/tune-pipeline --- # Tune a GlassFlow pipeline Move a pipeline from "barely keeping up" to "comfortably keeping up" without overprovisioning. Confirm the bottleneck from metrics, propose a specific change tied to a measured signal, apply it via the right endpoint, then re-read the metrics to verify the change helped. ## Prerequisites - A reachable GlassFlow API. Default `http://localhost:8081`; ask the user if remote. - Access to GlassFlow's emitted metrics. One of: - A Prometheus (or compatible) endpoint that scrapes the OTel Collector. The skill queries via `/api/v1/query` and `/api/v1/query_range`. - Direct access to the OTel Collector's `/metrics` endpoint at `{release}-otel-collector.{namespace}.svc.cluster.local:9090/metrics`. - A clear baseline: which signal is bad and by how much, captured *before* the change so the after-comparison is meaningful. ## Inputs to gather - `pipeline_id`: required. - `prometheus_endpoint`: required. - Symptom: required. One of (each maps to a specific metric in step 2): - Ingestor backpressure (gauge stays at 1, episodes lengthening) - NATS stream near capacity (stream_depth_ratio sustained > 0.8) - Sink lag (sink rate trails source rate) - DLQ filling under load (dlq_records_written_total rate elevated) - Sink retry exhaustion (sink_retries_total{outcome="exhausted"} > 0) - Target: required. Either a target throughput (events/sec), a target lag (seconds or messages), or "eliminate backpressure". - Optional constraints: max replicas, memory ceiling, storage ceiling. - `api_endpoint`: optional; default `http://localhost:8081`. - `metric_namespace`: optional; defaults to `glassflow`. ## Steps ### 1. Capture the baseline Snapshot the current state of both the configuration and the metrics that describe the symptom. You need a clean baseline so step 6 (verify) has something to compare against. ```bash ENDPOINT="" PID="" NS="" PROM="" # Config + resources curl -s "$ENDPOINT/api/v1/pipeline/$PID" > /tmp/tune-$PID-pipeline.json curl -s "$ENDPOINT/api/v1/pipeline/$PID/resources" > /tmp/tune-$PID-resources.json curl -s "$ENDPOINT/api/v1/pipeline/$PID/health" | python3 -m json.tool # Save the symptom metrics as a baseline (run all of these) { echo "=== source rate (events/sec, 5m) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=rate(${NS}_gfm_kafka_records_read_total{pipeline_id=\"$PID\"}[5m])" echo "=== sink rate (events/sec, 5m) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=rate(${NS}_gfm_clickhouse_records_written_total{pipeline_id=\"$PID\"}[5m])" echo "=== ingestor backpressure active ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=${NS}_gfm_ingestor_backpressure_active{pipeline_id=\"$PID\"}" echo "=== stream depth ratio ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=${NS}_gfm_stream_depth_ratio{pipeline_id=\"$PID\"}" echo "=== sink batch size records (p50, p95) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=histogram_quantile(0.5, sum by (le) (rate(${NS}_gfm_sink_batch_size_records_bucket{pipeline_id=\"$PID\"}[5m])))" curl -sG "$PROM/api/v1/query" --data-urlencode "query=histogram_quantile(0.95, sum by (le) (rate(${NS}_gfm_sink_batch_size_records_bucket{pipeline_id=\"$PID\"}[5m])))" echo "=== sink retries (outcome) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=sum by (outcome) (rate(${NS}_gfm_sink_retries_total{pipeline_id=\"$PID\"}[5m]))" } > /tmp/tune-$PID-before.txt ``` The pipeline must be `Running` for metrics to reflect live state. If it is `Failed` or `Stopped`, use the [`debug-pipeline`](../debug-pipeline/SKILL.md) skill first. ### 2. Map the symptom to a metric and a tunable Each row below names the canonical metric signature and the matching config knob. Pick the row that matches the user's symptom; do not propose changes without a matching metric. | Metric signature | Likely bottleneck | Tunable | |---|---|---| | `ingestor_backpressure_active == 1` sustained, `stream_depth_ratio` near 1.0 | Downstream consumer slower than the source; NATS stream filling | Scale the slowest downstream component (often sink replicas) or grow the stream (`resources.nats.stream.max_messages` / `.max_bytes`) | | `kafka_records_read_total` rate growing but consumer-group lag also growing, ingestor CPU low | Ingestor parallelism capped by partition count | Increase Kafka topic partitions (Kafka-side change, outside GlassFlow), then increase `resources.sources[].replicas` | | `kafka_records_read_total` rate flat near zero with non-zero topic lag | Ingestor under-provisioned per replica | Raise `resources.sources[].limits.cpu` and `.memory` | | `clickhouse_records_written_total` rate consistently below source rate; `sink_batch_size_records` p95 hitting the configured `max_batch_size` | Sink throttled by batch frequency / size | Tune `sink.max_batch_size` upward and/or `sink.max_delay_time` downward via `/edit`, or scale `resources.sink.replicas` | | `sink_retries_total{outcome="retry"}` rate > 0 sustained | Sink hitting retryable CH errors (overload, connection drops) | Raise `resources.sink.replicas`; check ClickHouse capacity separately | | `sink_retries_total{outcome="exhausted"}` rate > 0 | Sink giving up on batches; data loss risk | Out of tuning scope: surface to the user and run `debug-pipeline` to find the root cause | | `processing_duration_seconds` bucket high on `stage=dedup_*` with growing PVC | Dedup window too long for storage budget | Shrink dedup `time_window` (via `/edit`) or grow `resources.transform[].storage.size` | | `sink_errors_by_classification_total{classification="terminal"}` > 0 | Schema or type mismatch on the destination table | Out of tuning scope: this is a config problem, not a sizing problem | If no row matches, do not guess. Re-run `debug-pipeline` or ask the user to narrow the signal. ### 3. Propose a concrete diff Build a specific change with rationale tied to a measured value. Example to show the user: > Bottleneck: sink rate is 5,200 events/sec while source rate is 14,800 events/sec (gap = 9,600). > `sink_batch_size_records` p95 is at 5,000 (matches the configured `max_batch_size`), so the sink is batch-bound, not throughput-bound on individual inserts. > > Proposed change: > - `sink.max_batch_size`: 5000 -> 20000 (let the sink form larger batches; reduces per-insert overhead) > - `sink.max_delay_time`: 5s -> 2s (flush sooner so larger batches do not wait too long) > - `resources.sink.replicas`: 1 -> 2 (parallelize inserts; ClickHouse accepts concurrent INSERTs to the same table) > > Expected effect: sink rate climbs to >= 14,000 events/sec within ~5 minutes, `stream_depth_ratio` falls below 0.5, `ingestor_backpressure_active` returns to 0. > > Apply via: stop pipeline, PUT /resources for the replica change, POST /edit for the batch tuning, restart. Wait for the user to confirm before submitting any change. ### 4. Apply the change Choose the endpoint by what is changing. **Resource changes (replicas, requests/limits, storage, NATS stream sizing):** require the pipeline to be `Stopped`, then `PUT /api/v1/pipeline//resources`. ```bash # Stop curl -s -o /tmp/tune-stop.json -w "%{http_code}\n" -X POST "$ENDPOINT/api/v1/pipeline/$PID/stop" # Wait for Stopped until [ "$(curl -s "$ENDPOINT/api/v1/pipeline/$PID/health" | python3 -c "import sys,json; print(json.load(sys.stdin).get('overall_status'))")" = "Stopped" ]; do sleep 2; done # Apply resource diff (full PipelineResources body required) curl -s -o /tmp/tune-put.json -w "%{http_code}\n" -X PUT "$ENDPOINT/api/v1/pipeline/$PID/resources" \ -H 'Content-Type: application/json' \ --data @/tmp/tune-$PID-new-resources.json # Resume curl -s -o /tmp/tune-resume.json -w "%{http_code}\n" -X POST "$ENDPOINT/api/v1/pipeline/$PID/resume" ``` **Config changes (batch size, delay, dedup window, transforms, filter):** use `POST /api/v1/pipeline//edit` with the full new pipeline JSON. `/edit` applies the config and resumes the pipeline in one call. ```bash curl -s -o /tmp/tune-edit.json -w "%{http_code}\n" -X POST "$ENDPOINT/api/v1/pipeline/$PID/edit" \ -H 'Content-Type: application/json' \ --data @/tmp/tune-$PID-new-pipeline.json ``` Mixed changes (resources + config) require both calls in sequence: stop, PUT /resources, then `/edit` (which resumes) or `/resume`. ### 5. Wait for the pipeline to be `Running` again ```bash for i in {1..60}; do STATUS=$(curl -s "$ENDPOINT/api/v1/pipeline/$PID/health" | python3 -c "import sys,json; print(json.load(sys.stdin).get('overall_status','?'))") echo " attempt $i: overall_status=$STATUS" case "$STATUS" in Running) echo " ready"; break ;; Failed) echo " pipeline failed after change; revert"; exit 1 ;; Stopping|Stopped|Terminating) echo " pipeline did not resume; investigate"; exit 1 ;; esac sleep 5 done ``` ### 6. Verify the change helped (metrics, not assumptions) Wait at least 5 minutes after the pipeline returns to `Running` so the rate windows refill, then re-pull the same baseline queries from step 1: ```bash { echo "=== source rate (events/sec, 5m) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=rate(${NS}_gfm_kafka_records_read_total{pipeline_id=\"$PID\"}[5m])" echo "=== sink rate (events/sec, 5m) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=rate(${NS}_gfm_clickhouse_records_written_total{pipeline_id=\"$PID\"}[5m])" echo "=== ingestor backpressure active ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=${NS}_gfm_ingestor_backpressure_active{pipeline_id=\"$PID\"}" echo "=== stream depth ratio ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=${NS}_gfm_stream_depth_ratio{pipeline_id=\"$PID\"}" echo "=== sink batch size records (p50, p95) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=histogram_quantile(0.5, sum by (le) (rate(${NS}_gfm_sink_batch_size_records_bucket{pipeline_id=\"$PID\"}[5m])))" curl -sG "$PROM/api/v1/query" --data-urlencode "query=histogram_quantile(0.95, sum by (le) (rate(${NS}_gfm_sink_batch_size_records_bucket{pipeline_id=\"$PID\"}[5m])))" echo "=== sink retries (outcome) ===" curl -sG "$PROM/api/v1/query" --data-urlencode "query=sum by (outcome) (rate(${NS}_gfm_sink_retries_total{pipeline_id=\"$PID\"}[5m]))" } > /tmp/tune-$PID-after.txt ``` Compare `before` and `after`, then report: ``` Change applied: Observation window: 5 minutes Symptom metric: Before: After: Other signals: source rate: -> sink rate: -> stream_depth_ratio: -> ingestor BP active: -> Verdict: Next move: ``` If the metric improved as predicted, accept. If it did not, revert the change before trying anything else. Stacking unverified changes hides which one helped or hurt. ## Common pitfalls - **`resources.nats.stream.max_age` and `.max_bytes` are immutable after pipeline creation.** Changing them requires a new pipeline. Other NATS-stream fields are mutable. Surface this constraint when the proposed change involves these fields. - **Resource updates require the pipeline to be `Stopped`.** `PUT /resources` returns 409 with a `current_status` error if you forget. Always stop first. - **`/edit` requires `Running` (or the appropriate transition target).** It is not a universal "apply config" endpoint; it has lifecycle preconditions. - **More replicas does not always help.** Increasing ingestor replicas beyond the Kafka topic's partition count buys nothing; replicas idle on partitions they cannot claim. Check partition count before suggesting parallelism increases. - **Increasing `sink.max_batch_size` shifts memory pressure to the sink Pods.** If the sink limit is `memory: 512Mi`, a 50k batch with heavy rows may OOM. Raise `limits.memory` alongside. - **Dedup window changes do not retroactively re-emit duplicates.** Shrinking the window only affects records evaluated after the change lands; it does not flush state already in BadgerDB. - **Counters are cumulative; always wrap in `rate()` over the same window for before/after comparison.** Comparing raw counter values across the change is meaningless. - **`{namespace}` in metric names is the Helm release namespace, not the per-pipeline one.** A non-default install changes every metric name. - **Changes are not free.** Stopping a pipeline incurs reconciliation time (10-60s). Doing many small tuning iterations is expensive. Batch related changes when you can.