e0b2114008
Files changed: .obsidian/workspace.json runbooks/binary-wrapper-interception.md runbooks/gpu-cpu-fallback-routing.md runbooks/pipeline-patterns.md runbooks/pipeline-probe-gate.md
726 lines
29 KiB
Markdown
726 lines
29 KiB
Markdown
# Pipeline & Wrapper Patterns
|
||
|
||
Three composable patterns for adding pre-flight logic to tools and pipelines you don't fully control:
|
||
|
||
1. **Binary Wrapper Interception** — the *delivery mechanism*. Transparently replace a CLI binary with a wrapper that runs custom logic, then `exec`s the real binary. The caller (service, cron, runner) never knows.
|
||
2. **GPU/CPU Fallback Routing** — *logic you inject*. Probe a job, route small→GPU / large→CPU, and gate concurrency with `flock` so excess jobs fail-fast and re-queue instead of OOM-killing each other.
|
||
3. **Pre-Flight Probe Gate** — *logic you inject*. Cheaply inspect each input and skip the expensive step when the work would be wasted (wrong format, already optimized, corrupt, too large).
|
||
|
||
They compose: **Pattern 1 is how you deploy; Patterns 2 and 3 are two kinds of pre-flight logic you put inside the wrapper.** The running example throughout is the Whisper auto-captioning / PeerTube transcoder / RECON extraction stack on cortex.
|
||
|
||
## Contents
|
||
|
||
- [Pattern 1 — Binary Wrapper Interception](#pattern-1--binary-wrapper-interception)
|
||
- [Pattern 2 — GPU/CPU Fallback Routing](#pattern-2--gpucpu-fallback-routing)
|
||
- [Pattern 3 — Pre-Flight Probe Gate](#pattern-3--pre-flight-probe-gate)
|
||
|
||
---
|
||
|
||
# Pattern 1 — Binary Wrapper Interception
|
||
|
||
Transparently intercept a CLI binary with a wrapper script that adds pre-flight logic (routing, validation, logging) before exec-ing the real binary. The caller — whether a service, cron job, or another script — never knows the difference.
|
||
|
||
Use this when you need to modify the behavior of a tool that's called by a system you don't control (e.g., a runner, scheduler, or third-party service), without changing the caller's config or code.
|
||
|
||
### Prerequisites
|
||
|
||
- The binary to intercept is installed and working
|
||
- You have root/sudo access on the target machine
|
||
- The caller invokes the binary by absolute path or via PATH lookup
|
||
|
||
### Inputs
|
||
|
||
Prompt the user for all of these before executing:
|
||
|
||
```
|
||
TARGET_HOST= # SSH alias or IP (e.g., cortex). Use "localhost" if local.
|
||
BINARY_NAME= # Name of the binary to intercept (e.g., "whisper-ctranslate2")
|
||
BINARY_PATH= # Full path to the binary (e.g., "/usr/local/bin/whisper-ctranslate2")
|
||
WRAPPER_NAME= # Name for the wrapper script (e.g., "whisper-smart")
|
||
WRAPPER_DIR= # Directory for the wrapper (e.g., "/usr/local/bin")
|
||
REAL_SUFFIX= # Suffix for the renamed real binary (default: "-real")
|
||
```
|
||
|
||
### Step 1: Locate the Real Binary
|
||
|
||
Find the actual binary or symlink that will be intercepted.
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "ls -la $BINARY_PATH && file $BINARY_PATH"
|
||
```
|
||
|
||
If it's already a symlink, follow it to the real target:
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "readlink -f $BINARY_PATH"
|
||
```
|
||
|
||
**Gate:** Must return a valid file. Record the real binary location — you'll need it for the rename.
|
||
|
||
### Step 2: Rename the Real Binary
|
||
|
||
Move the original binary out of the way so the wrapper can take its place.
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "sudo mv $BINARY_PATH ${BINARY_PATH}${REAL_SUFFIX}"
|
||
```
|
||
|
||
If the original was a symlink (e.g., pip-installed Python tool):
|
||
|
||
```bash
|
||
# Preserve the symlink target
|
||
REAL_TARGET=$(ssh $TARGET_HOST "readlink -f $BINARY_PATH")
|
||
ssh $TARGET_HOST "sudo rm $BINARY_PATH && sudo ln -s $REAL_TARGET ${BINARY_PATH}${REAL_SUFFIX}"
|
||
```
|
||
|
||
**Gate:**
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "ls -la ${BINARY_PATH}${REAL_SUFFIX}"
|
||
ssh $TARGET_HOST "${BINARY_PATH}${REAL_SUFFIX} --version 2>/dev/null || ${BINARY_PATH}${REAL_SUFFIX} --help 2>/dev/null | head -1"
|
||
```
|
||
|
||
The renamed binary must exist and be executable. If not, undo immediately:
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "sudo mv ${BINARY_PATH}${REAL_SUFFIX} $BINARY_PATH"
|
||
```
|
||
|
||
### Step 3: Write the Wrapper Script
|
||
|
||
Create the wrapper at `$WRAPPER_DIR/$WRAPPER_NAME`. The wrapper must:
|
||
|
||
1. Accept all original arguments (`$@`)
|
||
2. Perform pre-flight logic (inspection, routing, logging)
|
||
3. `exec` the real binary with (possibly modified) arguments
|
||
4. Never silently swallow errors — if pre-flight fails, exit with a meaningful code
|
||
|
||
Template:
|
||
|
||
```bash
|
||
#!/bin/bash
|
||
# Wrapper for $BINARY_NAME — transparently intercepts calls
|
||
# Real binary at: ${BINARY_PATH}${REAL_SUFFIX}
|
||
|
||
LOGFILE="/tmp/${WRAPPER_NAME}.log"
|
||
|
||
# ──── Pre-flight logic ────
|
||
# Add your inspection, routing, or validation here.
|
||
# Example: inspect input files, check resource availability, choose parameters.
|
||
|
||
# Parse arguments to find relevant inputs (file paths, flags, etc.)
|
||
# This section is use-case specific.
|
||
|
||
# ──── Logging ────
|
||
echo "[WRAPPER] $(date) args: $@" >> "$LOGFILE"
|
||
|
||
# ──── Execute real binary ────
|
||
# Use exec to replace this process — caller sees the real binary's exit code,
|
||
# stdout, stderr, and signal handling as if wrapper didn't exist.
|
||
exec ${BINARY_PATH}${REAL_SUFFIX} "$@"
|
||
```
|
||
|
||
Deploy the wrapper:
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "sudo tee $WRAPPER_DIR/$WRAPPER_NAME > /dev/null << 'WRAPPER'
|
||
<paste wrapper script here>
|
||
WRAPPER
|
||
sudo chmod +x $WRAPPER_DIR/$WRAPPER_NAME"
|
||
```
|
||
|
||
**Gate:**
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "ls -la $WRAPPER_DIR/$WRAPPER_NAME && head -1 $WRAPPER_DIR/$WRAPPER_NAME"
|
||
```
|
||
|
||
Must show executable permissions and `#!/bin/bash` shebang.
|
||
|
||
### Step 4: Install the Symlink
|
||
|
||
Replace the original binary path with a symlink to the wrapper.
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "sudo ln -sf $WRAPPER_DIR/$WRAPPER_NAME $BINARY_PATH"
|
||
```
|
||
|
||
**Gate:**
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "ls -la $BINARY_PATH && ls -la ${BINARY_PATH}${REAL_SUFFIX}"
|
||
```
|
||
|
||
Should show:
|
||
```
|
||
BINARY_PATH -> WRAPPER_DIR/WRAPPER_NAME (wrapper)
|
||
BINARY_PATH-real -> /path/to/actual/binary (real binary)
|
||
```
|
||
|
||
### Step 5: Test the Interception
|
||
|
||
Run the binary as the caller would. The wrapper should intercept transparently.
|
||
|
||
```bash
|
||
# Direct invocation
|
||
ssh $TARGET_HOST "$BINARY_PATH --version"
|
||
# Check wrapper log
|
||
ssh $TARGET_HOST "tail -5 /tmp/${WRAPPER_NAME}.log"
|
||
```
|
||
|
||
The `--version` output should come from the real binary. The log should show the wrapper fired. Test with actual workload arguments:
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "$BINARY_PATH <typical args here>"
|
||
ssh $TARGET_HOST "tail -1 /tmp/${WRAPPER_NAME}.log"
|
||
```
|
||
|
||
**Gate:** Both must succeed. If the binary fails or produces different output than before, the wrapper has a bug — check argument passing (quoting, `$@` vs `$*`).
|
||
|
||
### Step 6: Verify Service Integration
|
||
|
||
If the binary is called by a service (systemd, cron, etc.), restart that service and confirm it picks up the wrapper.
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "sudo systemctl restart <service-name>"
|
||
ssh $TARGET_HOST "sleep 5 && tail -5 /tmp/${WRAPPER_NAME}.log"
|
||
```
|
||
|
||
The log should show entries from the service's invocations, not just your manual tests.
|
||
|
||
### Rollback
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "sudo rm $BINARY_PATH && sudo mv ${BINARY_PATH}${REAL_SUFFIX} $BINARY_PATH"
|
||
# Or if the original was a symlink:
|
||
ssh $TARGET_HOST "sudo rm $BINARY_PATH && sudo ln -s <original-target> $BINARY_PATH"
|
||
```
|
||
|
||
No service restart needed — next invocation hits the real binary directly.
|
||
|
||
### Key Principles
|
||
|
||
1. **`exec` is mandatory.** Without `exec`, the wrapper runs the binary as a child process, which breaks signal handling (SIGTERM won't reach the real binary) and doubles PID usage. `exec` replaces the wrapper process entirely.
|
||
2. **Use `"$@"` not `$@` or `$*`.** Quoted `"$@"` preserves argument boundaries. Unquoted `$@` splits arguments with spaces. `$*` merges all arguments into one string.
|
||
3. **Appended flags override earlier ones.** Many CLI tools (argparse, getopt) use last-value-wins for duplicate flags. The wrapper can append `--flag value` after `"$@"` to force overrides without removing the caller's original flags.
|
||
4. **Exit codes matter.** If pre-flight fails, exit with a non-zero code the caller understands. Some callers retry on specific exit codes (e.g., PeerTube runner retries on exit 1).
|
||
5. **Log to /tmp, not the service's log directory.** The wrapper log is a debug artifact, not part of the service's data. `/tmp` is cleaned on reboot, which is fine for wrapper logs.
|
||
|
||
### Troubleshooting
|
||
|
||
- **Wrapper not being called:** Check the symlink chain (`ls -la $BINARY_PATH`). If the service uses a hardcoded absolute path that bypasses PATH, the symlink might be in the wrong location.
|
||
- **Arguments with spaces break:** Use `"$@"` (quoted) in the exec line, not `$@` (unquoted).
|
||
- **Service fails after wrapper install:** Check the shebang (`#!/bin/bash`), permissions (`chmod +x`), and that `exec` is present.
|
||
- **Wrapper log is empty:** The service might be calling a different path. Check `which $BINARY_NAME` vs what the service config specifies.
|
||
|
||
### Example — Whisper transcription routing (PeerTube runner on cortex)
|
||
|
||
The PeerTube remote runner calls `whisper-ctranslate2` for auto-captioning. The smart wrapper intercepts this to route short videos to GPU and long videos to CPU (the routing logic itself is **Pattern 2** below).
|
||
|
||
```
|
||
BINARY_NAME=whisper-ctranslate2
|
||
BINARY_PATH=/usr/local/bin/whisper-ctranslate2
|
||
WRAPPER_NAME=whisper-smart
|
||
REAL_SUFFIX=-real
|
||
|
||
Symlink chain:
|
||
/usr/local/bin/whisper-ctranslate2 → /usr/local/bin/whisper-smart
|
||
/usr/local/bin/whisper-ctranslate2-real → /home/zvx/.local/bin/whisper-ctranslate2
|
||
|
||
Wrapper logic:
|
||
- ffprobe audio duration from first non-flag argument
|
||
- < 1hr → exec with --device cuda --compute_type float16
|
||
- >= 1hr → exec with --device cpu --compute_type int8
|
||
- Appends --model medium after $@ (last-value-wins override)
|
||
```
|
||
|
||
---
|
||
|
||
# Pattern 2 — GPU/CPU Fallback Routing
|
||
|
||
Route workloads to GPU or CPU based on pre-flight inspection of job properties (duration, file size, resolution, complexity). Small jobs go to GPU for speed; large jobs fall back to CPU to avoid VRAM exhaustion. Concurrent job control via `flock` prevents OOM kills — excess jobs fail fast and re-queue instead of competing for memory.
|
||
|
||
Use this when you have a GPU workload where some jobs exceed VRAM capacity, and the system needs to handle both small and large jobs without manual intervention or OOM kills. **Deploy it via Pattern 1** (the router script *is* the wrapper's pre-flight logic).
|
||
|
||
### Prerequisites
|
||
|
||
- NVIDIA GPU with working drivers (`nvidia-smi` returns output)
|
||
- Both GPU and CPU execution paths available for the workload
|
||
- A probe tool to inspect job properties before execution (e.g., `ffprobe`, `mediainfo`, `file`, `wc`)
|
||
- A caller that retries on non-zero exit codes (scheduler, job queue, runner)
|
||
|
||
### Inputs
|
||
|
||
```
|
||
TARGET_HOST= # Machine with GPU (e.g., cortex)
|
||
WORKLOAD_BINARY= # The tool that processes jobs (e.g., "whisper-ctranslate2-real")
|
||
PROBE_TOOL= # Tool to inspect job properties (e.g., "ffprobe", "mediainfo")
|
||
GPU_VRAM_MB= # Total VRAM available (e.g., 16384 for 16GB)
|
||
WORKLOAD_VRAM_MB= # VRAM used per GPU job (e.g., 3700)
|
||
WORKLOAD_RAM_MB= # RAM used per CPU job (e.g., 11000)
|
||
THRESHOLD_VALUE= # Cutoff for GPU vs CPU routing (e.g., 3600 for seconds)
|
||
THRESHOLD_UNIT= # What the threshold measures (e.g., "seconds", "bytes", "pixels")
|
||
MAX_GPU_JOBS= # Max concurrent GPU jobs (e.g., 2)
|
||
MAX_CPU_JOBS= # Max concurrent CPU jobs (e.g., 1)
|
||
GPU_ARGS= # Arguments for GPU execution (e.g., "--device cuda --compute_type float16")
|
||
CPU_ARGS= # Arguments for CPU execution (e.g., "--device cpu --compute_type int8")
|
||
```
|
||
|
||
### Step 1: Determine the Routing Threshold
|
||
|
||
Profile representative workloads to find the VRAM crossover point.
|
||
|
||
```bash
|
||
# Run a small job on GPU, monitor VRAM
|
||
ssh $TARGET_HOST "nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits"
|
||
# Run the workload... check peak VRAM during execution
|
||
# Run a large job on GPU, watch for OOM — if it exceeds VRAM, that's your upper bound
|
||
```
|
||
|
||
Set the threshold conservatively below the point where GPU jobs start failing. Common strategies:
|
||
|
||
| Workload Type | Probe Property | Typical Threshold |
|
||
|---------------|----------------|-------------------|
|
||
| Audio transcription | Duration (seconds) | 1-2 hours |
|
||
| Image generation | Resolution (megapixels) | Based on model VRAM curve |
|
||
| Video encoding | Duration × resolution | Derived from VRAM budget |
|
||
| LLM inference | Token count / context length | Model-specific |
|
||
|
||
**Gate:** You must have a clear, measurable property that predicts VRAM usage. If the relationship is unpredictable, this pattern won't work — use a different strategy (e.g., try GPU first, fall back on OOM).
|
||
|
||
### Step 2: Write the Probe Function
|
||
|
||
```bash
|
||
# Generic probe template
|
||
probe_workload() {
|
||
local INPUT="$1"
|
||
local METRIC=0
|
||
|
||
if [[ -n "$INPUT" && -f "$INPUT" ]]; then
|
||
# Example: audio/video duration via ffprobe
|
||
METRIC=$($PROBE_TOOL -v quiet -show_entries format=duration \
|
||
-of csv=p=0 "$INPUT" 2>/dev/null | cut -d. -f1)
|
||
METRIC=${METRIC:-0}
|
||
|
||
# Example: file size in bytes
|
||
# METRIC=$(stat -c%s "$INPUT" 2>/dev/null)
|
||
|
||
# Example: image resolution (width × height)
|
||
# METRIC=$($PROBE_TOOL -v quiet -show_entries stream=width,height \
|
||
# -of csv=p=0 "$INPUT" 2>/dev/null | awk -F, '{print $1*$2}')
|
||
fi
|
||
|
||
echo "$METRIC"
|
||
}
|
||
```
|
||
|
||
**Gate:** Test the probe against known inputs:
|
||
|
||
```bash
|
||
probe_workload /path/to/small/input # Should be < THRESHOLD_VALUE
|
||
probe_workload /path/to/large/input # Should be >= THRESHOLD_VALUE
|
||
```
|
||
|
||
### Step 3: Implement the Router
|
||
|
||
```bash
|
||
#!/bin/bash
|
||
# GPU/CPU Fallback Router — routes jobs based on $THRESHOLD_UNIT inspection
|
||
|
||
THRESHOLD=$THRESHOLD_VALUE
|
||
LOGFILE="/tmp/workload-router.log"
|
||
GPU_LOCK="/tmp/gpu-workload.lock"
|
||
CPU_LOCK="/tmp/cpu-workload.lock"
|
||
|
||
# ──── Probe ────
|
||
INPUT="<extract from $@>"
|
||
METRIC=$(probe_workload "$INPUT")
|
||
|
||
# ──── Route ────
|
||
if (( METRIC < THRESHOLD )); then
|
||
MODE="GPU"; DEVICE_ARGS="$GPU_ARGS"; LOCK_FILE="$GPU_LOCK"; MAX_CONCURRENT=$MAX_GPU_JOBS
|
||
else
|
||
MODE="CPU"; DEVICE_ARGS="$CPU_ARGS"; LOCK_FILE="$CPU_LOCK"; MAX_CONCURRENT=$MAX_CPU_JOBS
|
||
fi
|
||
|
||
# ──── Concurrency control ────
|
||
if (( MAX_CONCURRENT == 1 )); then
|
||
# Single-job lock: flock with fail-fast
|
||
exec 9>"$LOCK_FILE"
|
||
if ! flock --nonblock 9; then
|
||
echo "[ROUTER] $(date) mode=${MODE}-BLOCKED metric=${METRIC} (slot full, exiting)" >> "$LOGFILE"
|
||
exit 1 # Caller should retry later
|
||
fi
|
||
fi
|
||
# For MAX_CONCURRENT > 1, use numbered lock files:
|
||
# for i in $(seq 0 $((MAX_CONCURRENT - 1))); do
|
||
# SLOT_LOCK="${LOCK_FILE}.${i}"
|
||
# exec 9>"$SLOT_LOCK"
|
||
# if flock --nonblock 9; then break; fi # Got a slot
|
||
# if (( i == MAX_CONCURRENT - 1 )); then
|
||
# echo "[ROUTER] $(date) mode=${MODE}-BLOCKED metric=${METRIC} (all slots full)" >> "$LOGFILE"
|
||
# exit 1
|
||
# fi
|
||
# done
|
||
|
||
# ──── Log and execute ────
|
||
echo "[ROUTER] $(date) mode=$MODE metric=${METRIC} args: $@" >> "$LOGFILE"
|
||
exec $WORKLOAD_BINARY "$@" $DEVICE_ARGS
|
||
```
|
||
|
||
**Key design decisions:**
|
||
|
||
- **`flock --nonblock`**: Non-blocking lock attempt. If the slot is taken, exit immediately instead of waiting — this prevents queue starvation where all runner slots block waiting for CPU jobs.
|
||
- **Exit code 1**: The caller (runner, scheduler) should interpret this as "retry later." Most job queues do by default.
|
||
- **`exec`**: Replace the router process with the workload binary so signals, exit codes, and resource limits pass through cleanly.
|
||
- **Lock files in `/tmp`**: Auto-cleaned on reboot. No stale locks after crashes.
|
||
|
||
### Step 4: Integrate with the Caller
|
||
|
||
Deploy the router using **Pattern 1 (Binary Wrapper Interception)**:
|
||
|
||
1. Rename the real binary: `mv $BINARY → ${BINARY}-real`
|
||
2. Write the router script (above) as the wrapper
|
||
3. Symlink: `ln -sf /path/to/router $BINARY`
|
||
|
||
Or, if the caller supports configurable command paths, point it directly at the router.
|
||
|
||
### Step 5: Verify Both Paths
|
||
|
||
```bash
|
||
# GPU path — submit a small job, expect mode=GPU + GPU utilization
|
||
ssh $TARGET_HOST "$BINARY <small-input-args>"
|
||
ssh $TARGET_HOST "nvidia-smi --query-gpu=utilization.gpu,memory.used --format=csv,noheader"
|
||
ssh $TARGET_HOST "tail -1 /tmp/workload-router.log" # mode=GPU
|
||
|
||
# CPU path — submit a large job, expect mode=CPU + idle GPU
|
||
ssh $TARGET_HOST "$BINARY <large-input-args>"
|
||
ssh $TARGET_HOST "free -h"
|
||
ssh $TARGET_HOST "tail -1 /tmp/workload-router.log" # mode=CPU
|
||
|
||
# Concurrency — start one CPU job, immediately try a second
|
||
ssh $TARGET_HOST "$BINARY <large-input-1> &"; sleep 2
|
||
ssh $TARGET_HOST "$BINARY <large-input-2>" # second exits immediately, code 1
|
||
ssh $TARGET_HOST "grep BLOCKED /tmp/workload-router.log" # mode=CPU-BLOCKED
|
||
```
|
||
|
||
### Step 6: Tune and Monitor
|
||
|
||
```bash
|
||
ssh $TARGET_HOST "grep -c 'mode=GPU' /tmp/workload-router.log"
|
||
ssh $TARGET_HOST "grep -c 'mode=CPU' /tmp/workload-router.log"
|
||
ssh $TARGET_HOST "grep -c 'BLOCKED' /tmp/workload-router.log"
|
||
```
|
||
|
||
If BLOCKED count is high relative to CPU count, the threshold may be too aggressive (routing too many jobs to CPU). Raise the threshold or increase `MAX_CPU_JOBS` if RAM allows.
|
||
|
||
### Memory Budget Worksheet
|
||
|
||
```
|
||
GPU path:
|
||
VRAM per job: $WORKLOAD_VRAM_MB MB
|
||
Max GPU jobs: $MAX_GPU_JOBS
|
||
Total GPU VRAM: $GPU_VRAM_MB MB
|
||
Headroom: GPU_VRAM_MB - (WORKLOAD_VRAM_MB × MAX_GPU_JOBS) MB → must be positive
|
||
|
||
CPU path:
|
||
RAM per job: $WORKLOAD_RAM_MB MB
|
||
Max CPU jobs: $MAX_CPU_JOBS
|
||
System RAM: $(free -m | awk '/Mem:/{print $2}') MB
|
||
Other processes: ~2-4 GB (OS, services, buffers)
|
||
Headroom: SystemRAM - OtherProcs - (WORKLOAD_RAM_MB × MAX_CPU_JOBS) MB → must be positive
|
||
|
||
systemd MemoryMax: Set to MAX(GPU peak, CPU peak) + 20% buffer
|
||
```
|
||
|
||
### Troubleshooting
|
||
|
||
- **GPU job OOM-kills despite being under threshold:** Threshold too high, or VRAM varies by input beyond what the probe measures. Lower it or add a secondary probe (e.g., resolution in addition to duration).
|
||
- **CPU jobs pile up and exhaust RAM:** `MAX_CPU_JOBS` too high, or `flock` isn't working. Check lock files in `/tmp/` and the `exec 9>` redirect.
|
||
- **All jobs route to CPU:** The probe returns 0 or fails silently. Test it manually: `$PROBE_TOOL -v quiet -show_entries format=duration -of csv=p=0 /path/to/input`. Empty result usually means a permissions/path issue.
|
||
- **Blocked jobs never get retried:** The caller doesn't retry on exit 1. Match the exit code to what the caller expects (some want 75 for "temporary failure").
|
||
- **Lock files persist after crash:** `/tmp` clears on reboot, so locks self-heal. For immediate cleanup: `rm /tmp/cpu-workload.lock`.
|
||
|
||
### Example — Whisper auto-captioning on PeerTube runner (cortex)
|
||
|
||
```
|
||
WORKLOAD_BINARY=/usr/local/bin/whisper-ctranslate2-real
|
||
PROBE_TOOL=ffprobe
|
||
GPU_VRAM_MB=16384 # RTX A4000
|
||
WORKLOAD_VRAM_MB=3700 # Whisper medium on float16
|
||
WORKLOAD_RAM_MB=11000 # Whisper medium on CPU int8 (peak for 9.5hr video)
|
||
THRESHOLD_VALUE=3600 # 1 hour in seconds
|
||
THRESHOLD_UNIT=seconds
|
||
MAX_GPU_JOBS=2 # Runner concurrency=2, both can be GPU
|
||
MAX_CPU_JOBS=1 # Only 1 CPU job at a time (11GB peak, 20G MemoryMax)
|
||
GPU_ARGS="--device cuda --compute_type float16"
|
||
CPU_ARGS="--device cpu --compute_type int8"
|
||
|
||
Result: 4100+ videos captioned. ~20 videos over 1 hour routed to CPU.
|
||
GPU jobs: ~3.7GB VRAM, 88-99% GPU utilization. CPU jobs: ~8-11GB RAM, serialized via flock.
|
||
MemoryMax=20G on the runner service as safety net.
|
||
```
|
||
|
||
---
|
||
|
||
# Pattern 3 — Pre-Flight Probe Gate
|
||
|
||
Insert a cheap inspection step before expensive processing in a pipeline. Probe the input (ffprobe, mediainfo, file headers, checksums) to skip work that will be wasted — wrong format, already optimized, below quality threshold, or too large to process safely. Log every decision for an audit trail. Keep a post-processing safety net as backup.
|
||
|
||
Use this when your pipeline processes files in bulk and a significant percentage of inputs don't need the expensive step, or when processing the wrong input would waste time, storage, or GPU cycles. Like Pattern 2, the gate can live inside a Pattern 1 wrapper, or inline in a processing loop.
|
||
|
||
### Prerequisites
|
||
|
||
- A pipeline with at least one expensive processing step (transcoding, inference, embedding, etc.)
|
||
- A probe tool that can inspect inputs cheaply (< 1 second per file)
|
||
- Clear criteria for what constitutes a "skip" vs "process" decision
|
||
|
||
### Inputs
|
||
|
||
```
|
||
PIPELINE_NAME= # Human-readable name (e.g., "video-transcoder", "pdf-extractor")
|
||
PROBE_TOOL= # Inspection tool (e.g., "ffprobe", "mediainfo", "file", "pdfinfo")
|
||
INPUT_DIR= # Where the pipeline reads inputs (e.g., "/opt/pipeline/incoming")
|
||
OUTPUT_DIR= # Where processed outputs go (e.g., "/opt/pipeline/processed")
|
||
SKIP_DIR= # Where skipped inputs go (e.g., "/opt/pipeline/skipped")
|
||
FAIL_DIR= # Where failed inputs go (e.g., "/opt/pipeline/failed")
|
||
LOG_FILE= # Decision log path (e.g., "/opt/pipeline/logs/probe-gate.log")
|
||
```
|
||
|
||
### Step 1: Define Skip Criteria
|
||
|
||
Enumerate the conditions under which a file should skip the expensive step. Be specific — vague criteria lead to false positives.
|
||
|
||
| Check | Probe Command | Skip When |
|
||
|-------|---------------|-----------|
|
||
| Video codec | `ffprobe -show_entries stream=codec_name` | Already target codec (e.g., already HEVC) |
|
||
| Audio bitrate | `ffprobe -show_entries stream=bit_rate` | Below minimum quality threshold |
|
||
| Resolution | `ffprobe -show_entries stream=width,height` | Below minimum (e.g., < 360p) |
|
||
| Duration | `ffprobe -show_entries format=duration` | Exceeds safe processing limit |
|
||
| File size | `stat -c%s` | Zero bytes, or exceeds storage budget |
|
||
| PDF pages | `pdfinfo file.pdf \| grep Pages` | Too many pages for OCR budget |
|
||
| Image format | `file --mime-type` | Already target format |
|
||
| Container format | `ffprobe -show_entries format=format_name` | Unsupported container |
|
||
| Corruption | `ffprobe -v error` exit code | Non-zero = corrupt file |
|
||
| Existing output | `test -f $OUTPUT_DIR/$(basename)` | Output already exists (dedup) |
|
||
|
||
**Gate:** Write your criteria as a decision table:
|
||
|
||
```
|
||
Criterion 1: <property> <operator> <value> → SKIP (reason: "<why>")
|
||
Criterion 2: <property> <operator> <value> → SKIP (reason: "<why>")
|
||
Criterion 3: <property> not available → SKIP (reason: "probe failed")
|
||
Default: → PROCESS
|
||
```
|
||
|
||
### Step 2: Write the Probe Gate Function
|
||
|
||
```bash
|
||
#!/bin/bash
|
||
# Pre-flight probe gate for $PIPELINE_NAME
|
||
# Returns: 0 = process, 1 = skip, 2 = fail (corrupt/unreadable)
|
||
|
||
LOGFILE="$LOG_FILE"
|
||
|
||
probe_gate() {
|
||
local INPUT="$1"
|
||
local BASENAME=$(basename "$INPUT")
|
||
local TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')
|
||
|
||
# ──── Existence check ────
|
||
if [[ ! -f "$INPUT" ]]; then
|
||
echo "[$TIMESTAMP] FAIL $BASENAME reason=file_not_found" >> "$LOGFILE"
|
||
return 2
|
||
fi
|
||
|
||
# ──── Size check ────
|
||
local SIZE=$(stat -c%s "$INPUT" 2>/dev/null)
|
||
if (( SIZE == 0 )); then
|
||
echo "[$TIMESTAMP] SKIP $BASENAME reason=zero_bytes size=0" >> "$LOGFILE"
|
||
return 1
|
||
fi
|
||
|
||
# ──── Probe the input (adapt to your probe tool and criteria) ────
|
||
local PROBE_OUTPUT
|
||
PROBE_OUTPUT=$($PROBE_TOOL <probe-specific-flags> "$INPUT" 2>/dev/null)
|
||
local PROBE_EXIT=$?
|
||
if (( PROBE_EXIT != 0 )); then
|
||
echo "[$TIMESTAMP] FAIL $BASENAME reason=probe_failed exit=$PROBE_EXIT" >> "$LOGFILE"
|
||
return 2
|
||
fi
|
||
|
||
# ──── Apply skip criteria ────
|
||
local CODEC=$(echo "$PROBE_OUTPUT" | grep codec_name | head -1 | cut -d= -f2)
|
||
if [[ "$CODEC" == "hevc" ]]; then
|
||
echo "[$TIMESTAMP] SKIP $BASENAME reason=already_hevc codec=$CODEC" >> "$LOGFILE"
|
||
return 1
|
||
fi
|
||
|
||
local HEIGHT=$(echo "$PROBE_OUTPUT" | grep '^height=' | head -1 | cut -d= -f2)
|
||
if (( HEIGHT < 240 )); then
|
||
echo "[$TIMESTAMP] SKIP $BASENAME reason=below_min_resolution height=$HEIGHT" >> "$LOGFILE"
|
||
return 1
|
||
fi
|
||
|
||
# ──── Passed all checks ────
|
||
echo "[$TIMESTAMP] PASS $BASENAME codec=$CODEC height=${HEIGHT} size=$SIZE" >> "$LOGFILE"
|
||
return 0
|
||
}
|
||
```
|
||
|
||
**Key design decisions:**
|
||
|
||
- **Return codes**: 0 = process (matches shell "success" convention), 1 = skip, 2 = fail. Callers branch on `$?`.
|
||
- **Structured log lines**: Every decision logged with timestamp, verdict, filename, reason — parseable by grep/awk.
|
||
- **Probe errors = FAIL, not SKIP**: If the probe itself fails, the file might be corrupt — route to fail dir for inspection rather than silently skipping.
|
||
|
||
### Step 3: Integrate into the Pipeline
|
||
|
||
**Option A — Inline in processing loop:**
|
||
|
||
```bash
|
||
for INPUT in "$INPUT_DIR"/*; do
|
||
probe_gate "$INPUT"
|
||
case $? in
|
||
0) process_file "$INPUT"; mv "$INPUT" "$OUTPUT_DIR/" ;; # Expensive step
|
||
1) mv "$INPUT" "$SKIP_DIR/" ;; # Skipped — preserve for audit
|
||
2) mv "$INPUT" "$FAIL_DIR/" ;; # Failed probe — needs investigation
|
||
esac
|
||
done
|
||
```
|
||
|
||
**Option B — As a pre-filter in a Pattern 1 wrapper** (add the gate to the wrapper before exec):
|
||
|
||
```bash
|
||
probe_gate "$INPUT_FILE"
|
||
GATE_RESULT=$?
|
||
if (( GATE_RESULT == 1 )); then
|
||
echo "[WRAPPER] $(date) SKIPPED: $INPUT_FILE" >> "$LOGFILE"
|
||
exit 0 # Success — nothing to do
|
||
fi
|
||
if (( GATE_RESULT == 2 )); then
|
||
echo "[WRAPPER] $(date) PROBE FAILED: $INPUT_FILE" >> "$LOGFILE"
|
||
exit 1 # Error — caller should retry or alert
|
||
fi
|
||
exec $REAL_BINARY "$@" # Gate passed — proceed with expensive processing
|
||
```
|
||
|
||
**Option C — In a Python pipeline script:**
|
||
|
||
```python
|
||
import subprocess, shutil, os
|
||
|
||
def probe_gate(input_path: str) -> tuple[str, dict]:
|
||
"""Returns (verdict, metadata) where verdict is 'process', 'skip', or 'fail'."""
|
||
if not os.path.exists(input_path):
|
||
return 'fail', {'reason': 'file_not_found'}
|
||
|
||
size = os.path.getsize(input_path)
|
||
if size == 0:
|
||
return 'skip', {'reason': 'zero_bytes', 'size': 0}
|
||
|
||
result = subprocess.run(
|
||
['ffprobe', '-v', 'quiet', '-show_entries', 'stream=codec_name,height',
|
||
'-of', 'flat', input_path],
|
||
capture_output=True, text=True
|
||
)
|
||
if result.returncode != 0:
|
||
return 'fail', {'reason': 'probe_failed', 'exit': result.returncode}
|
||
|
||
# Parse and apply criteria...
|
||
return 'process', {'codec': codec, 'height': height, 'size': size}
|
||
```
|
||
|
||
### Step 4: Add Post-Processing Safety Net
|
||
|
||
The probe gate is the primary filter, but add a post-processing check as backup. This catches cases where the probe was wrong (e.g., file reported as H.264 but was actually corrupt).
|
||
|
||
```bash
|
||
post_process_check() {
|
||
local OUTPUT="$1"
|
||
local TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')
|
||
|
||
# Size gate: output should be at least 10% of input size
|
||
local INPUT_SIZE=$2
|
||
local OUTPUT_SIZE=$(stat -c%s "$OUTPUT" 2>/dev/null)
|
||
if (( OUTPUT_SIZE < INPUT_SIZE / 10 )); then
|
||
echo "[$TIMESTAMP] POST-FAIL $OUTPUT reason=output_too_small input=${INPUT_SIZE} output=${OUTPUT_SIZE}" >> "$LOGFILE"
|
||
return 1
|
||
fi
|
||
|
||
# Integrity check: verify output is valid
|
||
$PROBE_TOOL -v error "$OUTPUT" 2>/dev/null
|
||
if (( $? != 0 )); then
|
||
echo "[$TIMESTAMP] POST-FAIL $OUTPUT reason=output_corrupt" >> "$LOGFILE"
|
||
return 1
|
||
fi
|
||
|
||
echo "[$TIMESTAMP] POST-PASS $OUTPUT size=$OUTPUT_SIZE" >> "$LOGFILE"
|
||
return 0
|
||
}
|
||
```
|
||
|
||
### Step 5: Reporting
|
||
|
||
```bash
|
||
echo "=== Probe Gate Report ==="
|
||
echo "Processed: $(grep -c ' PASS ' $LOG_FILE)"
|
||
echo "Skipped: $(grep -c ' SKIP ' $LOG_FILE)"
|
||
echo "Failed: $(grep -c ' FAIL ' $LOG_FILE)"
|
||
|
||
echo "Skip reasons:"
|
||
grep ' SKIP ' $LOG_FILE | grep -oP 'reason=\S+' | sort | uniq -c | sort -rn
|
||
|
||
echo "Failed files:"
|
||
grep ' FAIL ' $LOG_FILE | tail -10
|
||
```
|
||
|
||
### Troubleshooting
|
||
|
||
- **Probe is slow (> 1s/file):** Some tools read more than necessary. For ffprobe, use `-analyzeduration 1000000 -probesize 1000000`. For large PDFs, `pdfinfo` beats opening the file in Python.
|
||
- **Probe reports wrong codec/format:** Container and stream codecs can mismatch. Probe the stream level: `ffprobe -v quiet -select_streams v:0 -show_entries stream=codec_name -of csv=p=0 "$INPUT"`.
|
||
- **Skipped files that should have been processed:** Review the skip log; lower the threshold or add exceptions. The skip directory preserves files for re-processing if criteria change.
|
||
- **Post-processing catches failures the probe missed:** The safety net working as intended. Investigate why the probe missed it; add a new criterion if the pattern is common.
|
||
|
||
### Example A — PeerTube H.265 transcoding pipeline (cortex)
|
||
|
||
```
|
||
PIPELINE_NAME=video-transcoder
|
||
PROBE_TOOL=ffprobe
|
||
INPUT_DIR=/opt/bulk-import/completed
|
||
OUTPUT_DIR=/opt/bulk-import/transcoded
|
||
SKIP_DIR=/opt/bulk-import/skipped
|
||
FAIL_DIR=/opt/bulk-import/failed
|
||
|
||
Probe criteria:
|
||
- codec_name == "hevc" → SKIP (already H.265)
|
||
- height < 240 → SKIP (too low quality to bother)
|
||
- duration == 0 → FAIL (corrupt or audio-only)
|
||
- probe exit != 0 → FAIL (unreadable)
|
||
|
||
Post-processing safety net:
|
||
- output size < 10% of input → FAIL (transcode produced garbage)
|
||
- ffprobe on output fails → FAIL (corrupt output)
|
||
|
||
Result: Saved ~15% of GPU cycles by skipping already-optimized files.
|
||
```
|
||
|
||
### Example B — PDF extraction pipeline (RECON on VM 1130)
|
||
|
||
```
|
||
PIPELINE_NAME=pdf-extractor
|
||
PROBE_TOOL=pdfinfo
|
||
INPUT_DIR=/mnt/library/incoming
|
||
OUTPUT_DIR=/opt/recon/extracted
|
||
|
||
Probe criteria:
|
||
- Pages > 500 → route to Gemini Vision (OCR too slow)
|
||
- Pages == 0 → FAIL (corrupt PDF)
|
||
- File size < 1KB → SKIP (empty/placeholder)
|
||
- Encrypted: yes → SKIP (can't extract without password)
|
||
- Already in SQLite status table → SKIP (dedup)
|
||
```
|
||
|
||
---
|
||
|
||
*Merged 2026-06-15 from `binary-wrapper-interception.md`, `gpu-cpu-fallback-routing.md`, and `pipeline-probe-gate.md` (originally written 2026-02-17).*
|