tutorial

Monitoring Google Cloud Dataflow with Vigilmon

Google Cloud Dataflow runs your stream and batch data processing pipelines — but stalled streaming jobs and silent batch failures need external monitoring beyond GCP's built-in alerts. Here's how to monitor Dataflow job health with Vigilmon.

Google Cloud Dataflow is a fully managed service for running Apache Beam pipelines at scale — both streaming (real-time event processing) and batch (large-scale data transforms). It's reliable, but streaming jobs can fall behind on event processing, batch jobs can fail partway through large datasets, and GCP's built-in alerting requires CloudMonitoring policies that teams often configure too broadly or not at all. Vigilmon gives you a focused external health check that confirms your Dataflow jobs are healthy without wading through the GCP console.

What You'll Set Up

  • Cron heartbeat monitors to confirm batch Dataflow jobs complete on schedule
  • HTTP endpoint monitor backed by a Cloud Run service for streaming job health
  • Watermark lag alerting for streaming pipelines
  • Alert channels for job failures and processing lag

Prerequisites

  • A Google Cloud project with at least one Dataflow job (batch or streaming)
  • gcloud CLI installed and authenticated
  • A free Vigilmon account

Step 1: Add a Heartbeat Monitor for Batch Dataflow Jobs

Batch Dataflow jobs have a defined start and end. If a batch job fails or never starts, there's no external signal. Set up a Vigilmon heartbeat to confirm each batch job completes on schedule.

  1. Log in to vigilmon.online and click Add Monitor.
  2. Select Cron Heartbeat as the monitor type.
  3. Name it after your job: e.g., dataflow-daily-aggregation.
  4. Set the expected ping interval to match your job's schedule (e.g., 1440 minutes for a daily batch run).
  5. Copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123.
  6. Click Save.

Step 2: Ping Vigilmon at the End of Each Batch Pipeline

Apache Beam pipelines are Java, Python, or Go programs. Add a Vigilmon ping at the end of the pipeline's successful completion path.

Python (Apache Beam)

import apache_beam as beam
import urllib.request

def run_pipeline():
    options = beam.options.pipeline_options.PipelineOptions(
        runner="DataflowRunner",
        project="my-gcp-project",
        region="us-central1",
        temp_location="gs://my-bucket/temp",
    )

    with beam.Pipeline(options=options) as p:
        (
            p
            | "ReadFromBigQuery" >> beam.io.ReadFromBigQuery(query="SELECT ...")
            | "Transform" >> beam.Map(my_transform_fn)
            | "WriteToBigQuery" >> beam.io.WriteToBigQuery("my_project:my_dataset.output_table")
        )

    # Pipeline context manager exits only after job completes successfully
    urllib.request.urlopen(
        "https://vigilmon.online/heartbeat/abc123",
        timeout=5
    )

if __name__ == "__main__":
    run_pipeline()

Java (Apache Beam)

import java.net.HttpURLConnection;
import java.net.URL;

public class MyPipeline {
    public static void main(String[] args) throws Exception {
        Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).create());

        // ... pipeline definition ...

        PipelineResult result = p.run();
        result.waitUntilFinish(); // blocks until job completes

        if (result.getState() == PipelineResult.State.DONE) {
            // Ping Vigilmon on success
            URL url = new URL("https://vigilmon.online/heartbeat/abc123");
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setConnectTimeout(5000);
            conn.getResponseCode();
            conn.disconnect();
        }
    }
}

The ping runs only after waitUntilFinish() returns with a DONE state — meaning the Dataflow job completed without errors.

Step 3: Monitor Streaming Job Health via Cloud Run

Streaming Dataflow jobs run continuously and don't "finish" — so a heartbeat-per-run doesn't apply. Instead, deploy a Cloud Run service that queries the Dataflow API for streaming job health and watermark lag:

import os
import requests
from flask import Flask, jsonify
from google.auth import default
from google.auth.transport.requests import Request

app = Flask(__name__)

PROJECT_ID = os.environ["GCP_PROJECT_ID"]
REGION = os.environ["DATAFLOW_REGION"]
LAG_THRESHOLD_SECONDS = 60

def get_access_token():
    creds, _ = default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
    creds.refresh(Request())
    return creds.token

@app.route("/health/dataflow")
def dataflow_health():
    token = get_access_token()
    headers = {"Authorization": f"Bearer {token}"}

    url = (
        f"https://dataflow.googleapis.com/v1b3/projects/{PROJECT_ID}"
        f"/locations/{REGION}/jobs"
        f"?filter=ACTIVE&view=JOB_VIEW_SUMMARY"
    )
    resp = requests.get(url, headers=headers, timeout=10)
    jobs = resp.json().get("jobs", [])

    streaming_jobs = [j for j in jobs if j.get("type") == "JOB_TYPE_STREAMING"]

    if not streaming_jobs:
        return jsonify({"status": "no_streaming_jobs"}), 503

    for job in streaming_jobs:
        state = job.get("currentState", "")
        if state not in ("JOB_STATE_RUNNING",):
            return jsonify({"status": "not_running", "job": job["name"], "state": state}), 503

    return jsonify({"status": "ok", "streaming_jobs": len(streaming_jobs)}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

Deploy to Cloud Run:

gcloud run deploy dataflow-health \
  --source . \
  --region us-central1 \
  --allow-unauthenticated \
  --set-env-vars GCP_PROJECT_ID=my-project,DATAFLOW_REGION=us-central1

Add a Vigilmon HTTP monitor:

  1. Click Add Monitor → HTTP / HTTPS.
  2. Enter the Cloud Run URL: https://dataflow-health-abc123-uc.a.run.app/health/dataflow.
  3. Set Expected HTTP status to 200.
  4. Set Check interval to 2 minutes.
  5. Click Save.

Step 4: Monitor Watermark Lag for Streaming Jobs

Watermark lag measures how far behind your streaming pipeline is from real-time event timestamps. A growing lag means your pipeline is falling behind its input. Extend the Cloud Run health endpoint to include lag:

def get_job_metrics(job_id, token):
    url = (
        f"https://dataflow.googleapis.com/v1b3/projects/{PROJECT_ID}"
        f"/locations/{REGION}/jobs/{job_id}/metrics"
    )
    resp = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=10)
    return resp.json().get("metrics", [])

@app.route("/health/dataflow/lag")
def dataflow_lag_health():
    token = get_access_token()
    # ... get streaming jobs ...
    for job in streaming_jobs:
        metrics = get_job_metrics(job["id"], token)
        for metric in metrics:
            if metric.get("name", {}).get("name") == "SystemLag":
                lag_ms = metric.get("scalar", 0)
                lag_seconds = lag_ms / 1000
                if lag_seconds > LAG_THRESHOLD_SECONDS:
                    return jsonify({
                        "status": "lagging",
                        "job": job["name"],
                        "lag_seconds": lag_seconds
                    }), 503
    return jsonify({"status": "ok"}), 200

Add a separate Vigilmon monitor pointing to /health/dataflow/lag. When lag exceeds 60 seconds, the endpoint returns 503 and Vigilmon fires an alert — giving you an early warning before downstream consumers notice missing events.

Step 5: Set Up a Scheduled Cron Ping for Streaming Jobs

For critical streaming jobs, also run a cron job that pings a separate Vigilmon heartbeat every few minutes — giving you a liveness signal even if the Cloud Run health endpoint has deployment issues:

#!/bin/bash
# /opt/scripts/ping-dataflow-liveness.sh
PROJECT="my-gcp-project"
REGION="us-central1"
HEARTBEAT_URL="https://vigilmon.online/heartbeat/streaming-abc456"

RUNNING=$(gcloud dataflow jobs list \
  --project="$PROJECT" \
  --region="$REGION" \
  --status=active \
  --format="value(id)" 2>/dev/null | wc -l)

if [ "$RUNNING" -gt 0 ]; then
  curl -s "$HEARTBEAT_URL"
fi

Add to cron:

*/5 * * * * /opt/scripts/ping-dataflow-liveness.sh

Set the Vigilmon heartbeat interval to 15 minutes to allow a few missed pings before alerting.

Step 6: Configure Alert Channels

  1. Go to Alert Channels in Vigilmon and add Slack, PagerDuty, or email.
  2. For streaming jobs processing payment events or user activity, set Consecutive failures before alert to 1 — lag grows quickly once a pipeline stalls.
  3. For batch jobs where a retry is acceptable, set it to 2.
  4. Use Maintenance windows during Dataflow job upgrades or intentional restarts:
curl -X POST https://vigilmon.online/api/maintenance \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{"monitor_id": "abc123", "duration_minutes": 15}'

Summary

| Monitor | Target | What It Catches | |---|---|---| | Cron heartbeat | Batch pipeline completion | Failed batch run, missed schedule | | HTTP endpoint | Cloud Run status service | Streaming job not in RUNNING state | | HTTP endpoint (lag) | Cloud Run lag health route | Watermark lag exceeding threshold | | Cron heartbeat | Streaming job liveness cron | Cloud Run health endpoint outage |

Google Cloud Dataflow handles the hard parts of distributed stream and batch processing. Vigilmon handles the part Dataflow doesn't: telling you from outside GCP when a job has stopped producing results, fallen behind on lag, or simply never started. A few monitors and a Web Activity ping turn a silent failure into a page.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles