Decodable is a fully managed Apache Flink streaming platform that lets data engineers write Flink SQL pipelines without operating a single Flink cluster. You define connections (Kafka, Kinesis, S3, databases, REST APIs), pipelines (Flink SQL transformations), and streams (internal managed message channels), and Decodable handles provisioning, scaling, checkpointing, and upgrades. But "managed" doesn't mean you're off the hook for operational monitoring. When a Decodable pipeline enters a DEGRADED state, an upstream connection goes DISCONNECTED, or your stream backlog starts growing, you need to know before downstream consumers notice stale data. Vigilmon gives you HTTP-based health checks, heartbeats, and alerting that cover the critical operational dimensions of your Decodable-based data stack.
What You'll Set Up
- Pipeline status monitoring (RUNNING vs. DEGRADED)
- Connection health checks (CONNECTED vs. DISCONNECTED)
- Stream backlog / offset lag alerting
- Pipeline throughput (rows/sec) monitoring
- Decodable management API health checks
- Data freshness lag monitoring
- Pipeline error rate alerting
- Checkpoint and restart health tracking
- Egress sink write success monitoring
- Compute unit quota consumption alerts
Prerequisites
- A Decodable account with active pipelines and connections
- Decodable API access token (from the Decodable console under Settings → API Tokens)
- A small monitoring sidecar or Lambda function to poll the Decodable API
- A free Vigilmon account
Why Monitor Decodable?
Decodable's managed model means you don't manage Flink clusters — but you still own the operational state of your pipelines and connections. A pipeline entering DEGRADED state can happen because of an upstream schema change, a connection timeout, or a SQL logic bug encountering a bad record. A DISCONNECTED Kafka connection silently stops all pipelines that consume from it. Stream backlog can grow for hours before anyone notices the downstream consumer is reading stale data. And Decodable's compute unit billing means an unexpected workload surge can push you against your plan limit, throttling pipelines. These are operational failures that require your team's attention — and Vigilmon surfaces them in real time.
Step 1: Monitor Pipeline Status
The Decodable REST API exposes pipeline status at:
GET https://api.decodable.co/v1alpha2/pipelines
Pipeline objects include a status field with values: RUNNING, DEGRADED, STOPPED, STARTING.
Write a lightweight probe that polls this API and returns 503 if any pipeline is not RUNNING or STOPPED (i.e., unexpectedly DEGRADED or stuck in STARTING):
#!/usr/bin/env python3
"""Decodable pipeline health probe."""
import os
import json
import requests
from http.server import BaseHTTPRequestHandler, HTTPServer
DECODABLE_TOKEN = os.environ["DECODABLE_API_TOKEN"]
DECODABLE_ACCOUNT = os.environ["DECODABLE_ACCOUNT"] # your account name/slug
BASE_URL = f"https://api.decodable.co/v1alpha2"
HEADERS = {"Authorization": f"Bearer {DECODABLE_TOKEN}"}
def get_pipeline_health():
resp = requests.get(f"{BASE_URL}/pipelines", headers=HEADERS, timeout=10)
resp.raise_for_status()
pipelines = resp.json().get("items", [])
degraded = [
{"name": p["name"], "status": p.get("status", {}).get("current", "UNKNOWN")}
for p in pipelines
if p.get("status", {}).get("current") == "DEGRADED"
]
return degraded, len(pipelines)
class PipelineHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/pipeline-health":
try:
degraded, total = get_pipeline_health()
status = 200 if not degraded else 503
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(
{"degraded": degraded, "total_pipelines": total}
).encode())
except Exception as e:
self.send_response(503)
self.end_headers()
self.wfile.write(json.dumps({"error": str(e)}).encode())
else:
self.send_response(404)
self.end_headers()
def log_message(self, format, *args):
pass
HTTPServer(("0.0.0.0", 8080), PipelineHandler).serve_forever()
Deploy this probe (Docker container, Lambda on a schedule, or a small EC2/GCE instance) and add a Vigilmon monitor:
- Click Add Monitor → HTTP / HTTPS.
- Enter
http://<probe-host>:8080/pipeline-health. - Set Expected HTTP status to
200. - Set Check interval to
1 minute. - Click Save.
A 503 means at least one pipeline is DEGRADED — alert immediately.
Step 2: Monitor Connection Health
Decodable connections (Kafka, Kinesis, S3, database, REST) are the I/O layer for your pipelines. A DISCONNECTED connection stalls all pipelines reading from or writing to it.
def get_connection_health():
resp = requests.get(f"{BASE_URL}/connections", headers=HEADERS, timeout=10)
resp.raise_for_status()
connections = resp.json().get("items", [])
disconnected = [
{"name": c["name"], "type": c.get("connector", ""), "status": c.get("status", {}).get("current", "UNKNOWN")}
for c in connections
if c.get("status", {}).get("current") == "DISCONNECTED"
]
return disconnected, len(connections)
Expose /connection-health on port 8081 (same probe pattern). Return 503 when any connection is DISCONNECTED.
In Vigilmon, add a monitor at http://<probe-host>:8081/connection-health. A DISCONNECTED Kafka source means no records are flowing into your pipelines — alert immediately.
Step 3: Monitor Stream Backlog
Decodable streams are internal managed Kafka-compatible channels connecting pipelines. Backlog (unprocessed message lag) accumulates when a downstream pipeline processes records more slowly than the upstream pipeline produces them.
Use the Decodable API's stream metrics endpoint:
GET https://api.decodable.co/v1alpha2/streams/{stream-id}/metrics
This returns metrics including offset_lag. Write a probe:
STREAM_IDS = os.environ.get("DECODABLE_STREAM_IDS", "").split(",")
BACKLOG_THRESHOLD = int(os.environ.get("BACKLOG_THRESHOLD", "100000"))
def get_stream_backlog_health():
overloaded = []
for stream_id in STREAM_IDS:
resp = requests.get(
f"{BASE_URL}/streams/{stream_id.strip()}/metrics",
headers=HEADERS, timeout=10
)
if resp.status_code != 200:
continue
metrics = resp.json()
lag = metrics.get("offset_lag", 0)
if lag > BACKLOG_THRESHOLD:
overloaded.append({"stream_id": stream_id, "lag": lag, "threshold": BACKLOG_THRESHOLD})
return overloaded
Expose /stream-backlog on port 8082. Return 503 when any stream's lag exceeds the threshold. Monitor with Vigilmon at 2-minute intervals.
Step 4: Monitor Pipeline Throughput
Pipeline throughput (rows/sec) dropping to zero is the clearest signal of a stalled pipeline. Query pipeline metrics:
GET https://api.decodable.co/v1alpha2/pipelines/{pipeline-id}/metrics
The response includes input_records_per_second and output_records_per_second.
PIPELINE_IDS = os.environ.get("DECODABLE_PIPELINE_IDS", "").split(",")
def get_throughput_health():
stalled = []
for pipeline_id in PIPELINE_IDS:
resp = requests.get(
f"{BASE_URL}/pipelines/{pipeline_id.strip()}/metrics",
headers=HEADERS, timeout=10
)
if resp.status_code != 200:
continue
metrics = resp.json()
output_rps = metrics.get("output_records_per_second", 0)
if output_rps == 0:
stalled.append({"pipeline_id": pipeline_id, "output_rps": output_rps})
return stalled
Expose /throughput-health on port 8083. Return 503 when any pipeline's output rows/sec is zero. Alert on 503 — a stalled pipeline is producing no output data.
Step 5: Monitor the Decodable API Health
Your monitoring probe (and your team's pipeline management operations) depend on the Decodable API being responsive. Test it directly:
def decodable_api_health():
try:
start = time.time()
resp = requests.get(f"{BASE_URL}/pipelines", headers=HEADERS, timeout=10)
latency_ms = (time.time() - start) * 1000
if resp.status_code == 200 and latency_ms < 5000:
return True, {"latency_ms": latency_ms}
return False, {"status_code": resp.status_code, "latency_ms": latency_ms}
except Exception as e:
return False, {"error": str(e)}
Expose /api-health on port 8084. Return 503 on HTTP errors or latency > 5 seconds. Monitor with Vigilmon. API unavailability blocks all pipeline management operations.
Step 6: Monitor Data Freshness
Data freshness is the time since the last record was processed through each pipeline. A fresh pipeline processes records within seconds of arrival; a stale pipeline may have a connection issue that isn't yet showing as DEGRADED.
FRESHNESS_SLA_SECONDS = int(os.environ.get("FRESHNESS_SLA_SECONDS", "300"))
def get_freshness_health():
stale = []
resp = requests.get(f"{BASE_URL}/pipelines", headers=HEADERS, timeout=10)
for p in resp.json().get("items", []):
pid = p["id"]
metrics_resp = requests.get(
f"{BASE_URL}/pipelines/{pid}/metrics",
headers=HEADERS, timeout=10
)
if metrics_resp.status_code != 200:
continue
metrics = metrics_resp.json()
last_record_ts = metrics.get("last_record_processed_at")
if last_record_ts:
import datetime
last = datetime.datetime.fromisoformat(last_record_ts.replace("Z", "+00:00"))
age_seconds = (datetime.datetime.now(datetime.timezone.utc) - last).total_seconds()
if age_seconds > FRESHNESS_SLA_SECONDS:
stale.append({"pipeline": p["name"], "age_seconds": age_seconds})
return stale
Expose /freshness-health on port 8085. Return 503 when any pipeline's last processed record is older than the SLA threshold (default 5 minutes). Monitor with Vigilmon at 2-minute intervals.
Step 7: Monitor Pipeline Error Rate
Processing errors from SQL pipeline logic (type cast failures, null violations, malformed JSON) accumulate silently until they cause a pipeline to DEGRADE. Track error rates via pipeline metrics:
ERROR_RATE_THRESHOLD = float(os.environ.get("ERROR_RATE_THRESHOLD", "0.01"))
def get_error_rate_health():
high_error_pipelines = []
resp = requests.get(f"{BASE_URL}/pipelines", headers=HEADERS, timeout=10)
for p in resp.json().get("items", []):
pid = p["id"]
metrics_resp = requests.get(f"{BASE_URL}/pipelines/{pid}/metrics", headers=HEADERS, timeout=10)
if metrics_resp.status_code != 200:
continue
m = metrics_resp.json()
input_rps = m.get("input_records_per_second", 0)
error_rps = m.get("error_records_per_second", 0)
if input_rps > 0:
rate = error_rps / input_rps
if rate > ERROR_RATE_THRESHOLD:
high_error_pipelines.append({"pipeline": p["name"], "error_rate": rate})
return high_error_pipelines
Expose /error-rate-health on port 8086. Return 503 when any pipeline error rate exceeds 1%.
Step 8: Monitor Checkpoint and Restart Health
Decodable's Flink engine checkpoints pipeline state for fault tolerance. Frequent restarts indicate an unstable pipeline. Monitor via pipeline metadata:
def get_checkpoint_health():
resp = requests.get(f"{BASE_URL}/pipelines", headers=HEADERS, timeout=10)
unstable = []
for p in resp.json().get("items", []):
# Check pipeline restart_count in metrics or annotations
annotations = p.get("metadata", {}).get("annotations", {})
restart_count = int(annotations.get("decodable.co/restart-count", "0"))
if restart_count > 5:
unstable.append({"pipeline": p["name"], "restart_count": restart_count})
return unstable
Expose /checkpoint-health on port 8087. Alert on 503 when any pipeline has restarted more than 5 times — investigate the root cause before it causes data loss.
Step 9: Monitor Egress Sink Health
Decodable writes processed results to egress connections (Kafka, S3, database). A sink write failure means output data is being dropped.
Monitor egress connection status (covered in Step 2), but add a more specific check for egress-only connections:
EGRESS_CONNECTION_NAMES = os.environ.get("EGRESS_CONNECTION_NAMES", "").split(",")
def get_egress_health():
resp = requests.get(f"{BASE_URL}/connections", headers=HEADERS, timeout=10)
connections = {c["name"]: c for c in resp.json().get("items", [])}
failed = []
for name in EGRESS_CONNECTION_NAMES:
name = name.strip()
if name in connections:
status = connections[name].get("status", {}).get("current", "UNKNOWN")
if status != "CONNECTED":
failed.append({"name": name, "status": status})
return failed
Expose /egress-health on port 8088. A DISCONNECTED egress sink means pipeline output is being silently dropped — alert immediately.
Step 10: Monitor Compute Unit Quota
Decodable charges by compute unit (CU) consumption. Approaching or hitting your plan limit causes pipeline throttling. Query the usage API:
QUOTA_WARN_PCT = float(os.environ.get("QUOTA_WARN_PCT", "0.85"))
def get_quota_health():
resp = requests.get(f"{BASE_URL}/account/usage", headers=HEADERS, timeout=10)
if resp.status_code != 200:
return False, {"error": "could not fetch usage"}
usage = resp.json()
used = usage.get("compute_units_used", 0)
limit = usage.get("compute_units_limit", 1)
pct = used / limit if limit > 0 else 1.0
healthy = pct < QUOTA_WARN_PCT
return healthy, {"used_cu": used, "limit_cu": limit, "used_pct": round(pct * 100, 1)}
Expose /quota-health on port 8089. Return 503 when usage exceeds 85% of plan limit. Monitor with Vigilmon to get advance warning before throttling hits.
Alerting Configuration
| Monitor | Condition | Action |
|---------|-----------|--------|
| Pipeline status | Any 503 (pipeline DEGRADED) | Page on-call immediately |
| Connection health | Any 503 (connection DISCONNECTED) | Page on-call immediately |
| Stream backlog | 503 (lag > threshold) | Notify data engineering team |
| Pipeline throughput | 503 (output rps = 0) | Page on-call immediately |
| Decodable API health | Any 503 | Notify platform team |
| Data freshness | 503 (lag > SLA) | Notify data engineering team |
| Pipeline error rate | 503 (error rate > 1%) | Notify data engineering team |
| Checkpoint / restarts | 503 (restarts > 5) | Notify data engineering team |
| Egress sink health | Any 503 | Page on-call immediately |
| Compute unit quota | 503 (usage > 85%) | Notify data engineering team |
Configure notification channels in Vigilmon's Alerting settings to route to Slack, PagerDuty, or email.
Conclusion
Decodable's managed Flink service removes the burden of cluster operations, but your team still owns the operational health of pipelines, connections, and streams. By deploying a lightweight Python probe that polls the Decodable REST API and exposing HTTP health endpoints to Vigilmon, you get real-time visibility into all ten critical operational dimensions: pipeline state, connection connectivity, stream backlog, throughput, API availability, data freshness, error rates, checkpoint stability, egress sink health, and compute unit consumption. With Vigilmon alerting wired up, your team catches problems within minutes rather than finding out from downstream consumers.
Get started free at vigilmon.online.