data-pipeline-monitor

name

data-pipeline-monitor

description

Track ETL and data pipeline jobs with success/failure status, duration tracking, heartbeat monitoring, and dependency visualization. Use when you need to monitor scheduled jobs, detect failures, track pipeline health over time, or visualize ETL step dependencies. Triggers include "pipeline monitoring", "job tracking", "ETL status", "cron job health", "heartbeat monitor", "pipeline failed", or any task involving monitoring data workflows.

data-pipeline-monitor

Track ETL and data pipeline jobs. Self-hosted. REST API.

When to use

• Monitoring scheduled ETL jobs and cron tasks
• Detecting silent failures in data pipelines
• Tracking run duration history and detecting regressions
• Visualizing job dependency graphs (DAG)
• Getting alerts when pipelines fail or miss their schedule
• Any workflow where jobs need to report start, progress, and completion

Prerequisites

1. A running data-pipeline-monitor server
2. An API key (created in Settings or via API)

Installation

# Run with Docker
docker compose up -d

# Or run locally
pnpm install && pnpm build && pnpm start

Quick Start

# 1. Create an API key
curl -s -X POST http://localhost:3000/api/keys \
  -H 'X-API-Key: your-bootstrap-key' \
  -d '{"label":"my-server"}'
# -> {"key":"dpm_abc123...","label":"my-server"}  (shown once)

# 2. Register a pipeline
curl -s -X POST http://localhost:3000/api/pipelines \
  -H 'X-API-Key: dpm_abc123...' \
  -H 'Content-Type: application/json' \
  -d '{"id":"daily-etl","schedule":"0 2 * * *","jobs":[{"id":"extract"},{"id":"load","depends_on":["extract"]}]}'

# 3. Report a run
RUN_ID=$(curl -s -X POST http://localhost:3000/api/jobs/extract/start \
  -H 'X-API-Key: dpm_abc123...' | jq -r .run_id)
# ... do work ...
curl -s -X POST http://localhost:3000/api/jobs/extract/complete \
  -H 'X-API-Key: dpm_abc123...' \
  -d "{\"run_id\":\"$RUN_ID\"}"

Core API

Register a pipeline

POST /api/pipelines
Body: { id, name?, schedule?, expected_duration_s?, heartbeat_timeout_s?, jobs[], tags? }

Job lifecycle

POST /api/jobs/:id/start        -> { run_id, started_at }
POST /api/jobs/:id/heartbeat    -> body: { run_id }
POST /api/jobs/:id/complete     -> body: { run_id }
POST /api/jobs/:id/fail         -> body: { run_id, error? }

Query data

GET  /api/pipelines             -> list all pipelines
GET  /api/pipelines/:id         -> single pipeline + DAG + latest run
GET  /api/history               -> paginated run history
GET  /api/metrics               -> success rates, duration percentiles, run counts
GET  /api/health                -> health check (no auth required)

Heartbeat Pattern

Jobs must send periodic heartbeats or they are marked failed automatically.

# Recommended: heartbeat every heartbeat_timeout/3 seconds
(while true; do
  curl -s -X POST http://localhost:3000/api/jobs/$JOB_ID/heartbeat \
    -H "X-API-Key: $API_KEY" -d "{\"run_id\":\"$RUN_ID\"}"
  sleep 60
done) & HB_PID=$!

# Do your work
./run-job.sh

# Clean up
kill $HB_PID 2>/dev/null

# Report result
if [ $? -eq 0 ]; then
  curl -s -X POST .../complete -d "{\"run_id\":\"$RUN_ID\"}"
else
  curl -s -X POST .../fail -d "{\"run_id\":\"$RUN_ID\",\"error\":\"Exit $?\"}"
fi

Job Dependency (DAG)

Jobs in a pipeline can declare dependencies. The monitor validates no cycles exist.

{
  "id": "analytics-rollup",
  "jobs": [
    { "id": "fetch-events" },
    { "id": "fetch-users" },
    { "id": "aggregate", "depends_on": ["fetch-events", "fetch-users"] },
    { "id": "write-stats", "depends_on": ["aggregate"] }
  ]
}

Alert Rules

POST /api/alerts
Body: {
  condition: "on_failure" | "missed_run" | "duration_exceeded",
  pipeline_id?: string,       (null = all pipelines)
  threshold_s?: number,       (required for duration_exceeded)
  channel: "email" | "webhook",
  target: string              (email address or webhook URL)
}

CLI Reference

Command	Description
dpm pipelines	List all pipelines with status
dpm pipeline <id>	Show pipeline detail
dpm history [pipeline-id]	Show run history
dpm keys	List API keys
dpm keys create --label <label>	Create API key
dpm keys revoke <id>	Revoke API key
dpm --help	Show help
dpm --version	Show version

Environment Variables

Variable	Description	Default
DPM_PORT	HTTP server port	3000
DPM_HOST	Bind address	127.0.0.1
DPM_DATA_DIR	Data directory	~/.dpm
DPM_API_KEY	Bootstrap key (created on first run)	-
DPM_ALERT_EMAIL_FROM	From address for email alerts	-
DPM_SMTP_HOST	SMTP server host	-
DPM_SMTP_PORT	SMTP server port	587
DPM_SMTP_USER	SMTP username	-
DPM_SMTP_PASS	SMTP password	-
DPM_LOG_LEVEL	debug / info / warn / error	info
DPM_RETENTION_DAYS	Run history retention	90
DPM_DEV	Dev mode: verbose logging, no auth (0/1)	0

Job Status Values

Status	Meaning
running	Job started, heartbeats being received
success	Job completed normally
failed	Job failed (explicit fail call or heartbeat timeout)
pending	Job registered but not yet started in this run group
missed	Scheduled run did not start within the missed_run_window
skipped	Job skipped because a dependency failed

Troubleshooting

"heartbeat timeout"

The job did not send a heartbeat within heartbeat_timeout_s seconds. Start heartbeats immediately after the start call and ensure the background loop does not get killed before the job finishes.

"depends_on contains unknown job ID"

The depends_on array references a job ID that does not exist in the same pipeline. Check spelling and ensure all referenced jobs are in the jobs array.

Alert not firing

Check that the alert rule is enabled in Settings. Check the alert log at GET /api/alerts/log. Verify SMTP settings if using email channel.

"SQLITE_LOCKED" errors

The server is configured without WAL mode. This should not happen in a normal install. Check that DPM_DATA_DIR is on a local disk, not NFS.