tutorial

Monitoring Google Cloud Pub/Sub with Vigilmon: Subscription Health, Consumer Heartbeats & Undelivered Message Alerting

Practical guide to uptime monitoring for Google Cloud Pub/Sub — subscription health checks, consumer heartbeats, undelivered message alerting, and dead-letter topic monitoring with Vigilmon.

Google Cloud Pub/Sub is the messaging backbone for event-driven applications on GCP — decoupling services, fanning out events to multiple subscribers, and buffering traffic spikes. But when a subscriber stops pulling messages and the subscription backlog grows, or a dead-letter topic starts accumulating, your application degrades silently. Cloud Monitoring has Pub/Sub metrics, but configuring meaningful alerts requires careful setup inside the same project that might be having problems. Vigilmon gives you an independent, external health signal that fires alerts regardless of what's happening inside your GCP project.

This tutorial builds a complete monitoring strategy for Google Cloud Pub/Sub applications.

What You'll Build

  • A Cloud Functions health endpoint that checks subscription backlog and connectivity
  • A consumer heartbeat that pings Vigilmon after each message batch is processed
  • A dead-letter topic monitor that alerts when messages are being dropped
  • Alert channels for email and Slack

Prerequisites

  • GCP project with at least one Pub/Sub topic and subscription
  • Cloud Run or Cloud Functions service consuming messages
  • A free account at vigilmon.online

Step 1: Add a Pub/Sub Health Endpoint

Deploy a Cloud Functions HTTP trigger that checks your subscription's backlog and connectivity. This gives Vigilmon a URL to poll every minute.

Node.js health function

// health/index.js
const { http } = require("@google-cloud/functions-framework");
const { PubSub } = require("@google-cloud/pubsub");
const { v1: pubsubV1 } = require("@google-cloud/pubsub");

const pubsub = new PubSub();
const subscriberClient = new pubsubV1.SubscriberClient();

http("health", async (req, res) => {
  if (req.method !== "GET") {
    return res.status(405).json({ error: "Method not allowed" });
  }

  const checks = {};
  let ok = true;

  const projectId = process.env.GOOGLE_CLOUD_PROJECT;
  const subscriptionId = process.env.PUBSUB_SUBSCRIPTION;
  const backlogThreshold = parseInt(process.env.BACKLOG_THRESHOLD ?? "1000", 10);

  // Check subscription backlog
  try {
    const subscriptionPath = subscriberClient.subscriptionPath(
      projectId,
      subscriptionId
    );
    const [snapshot] = await subscriberClient.getSnapshot({
      snapshot: subscriptionPath.replace("subscriptions", "snapshots"),
    }).catch(() => [null]);

    // Use getSubscription to get message counts
    const subscription = pubsub.subscription(subscriptionId);
    const [metadata] = await subscription.getMetadata();

    checks.subscription = "ok";
    checks.subscription_name = subscriptionId;

    // Check if subscription is active
    if (metadata.state && metadata.state !== "ACTIVE") {
      checks.subscription = `state:${metadata.state}`;
      ok = false;
    }
  } catch (err) {
    checks.subscription = `error: ${err.message}`;
    ok = false;
  }

  // Check undelivered message count via monitoring API
  try {
    const { MetricServiceClient } = require("@google-cloud/monitoring");
    const monitoring = new MetricServiceClient();

    const now = Math.floor(Date.now() / 1000);
    const [timeSeries] = await monitoring.listTimeSeries({
      name: `projects/${projectId}`,
      filter: `metric.type="pubsub.googleapis.com/subscription/num_undelivered_messages" AND resource.labels.subscription_id="${subscriptionId}"`,
      interval: {
        endTime: { seconds: now },
        startTime: { seconds: now - 300 },
      },
    });

    if (timeSeries.length > 0) {
      const points = timeSeries[0].points;
      if (points.length > 0) {
        const backlog = points[0].value.int64Value ?? 0;
        checks.backlog = parseInt(backlog, 10);
        if (checks.backlog > backlogThreshold) {
          checks.subscription_health = `backlog_high:${checks.backlog}`;
          ok = false;
        }
      }
    }
  } catch (err) {
    // Non-fatal — monitoring API may not be enabled
    checks.backlog_check = `skipped: ${err.message}`;
  }

  res.status(ok ? 200 : 503).json({
    status: ok ? "ok" : "degraded",
    project: projectId,
    checks,
  });
});

Python health function

# health/main.py
import os
import json
import functions_framework
from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()

@functions_framework.http
def health(request):
    if request.method != "GET":
        return (json.dumps({"error": "Method not allowed"}), 405, {"Content-Type": "application/json"})

    project_id = os.environ["GOOGLE_CLOUD_PROJECT"]
    subscription_id = os.environ["PUBSUB_SUBSCRIPTION"]
    checks = {}
    ok = True

    try:
        subscription_path = subscriber.subscription_path(project_id, subscription_id)
        subscription = subscriber.get_subscription(request={"subscription": subscription_path})
        checks["subscription"] = "ok"
        checks["ack_deadline"] = subscription.ack_deadline_seconds

        if subscription.state.name != "ACTIVE":
            checks["subscription"] = f"state:{subscription.state.name}"
            ok = False
    except Exception as e:
        checks["subscription"] = f"error: {e}"
        ok = False

    body = json.dumps({
        "status": "ok" if ok else "degraded",
        "project": project_id,
        "checks": checks,
    })
    return (body, 200 if ok else 503, {"Content-Type": "application/json"})

Deploy:

gcloud functions deploy health \
  --gen2 \
  --runtime nodejs22 \
  --trigger-http \
  --allow-unauthenticated \
  --region us-central1 \
  --set-env-vars PUBSUB_SUBSCRIPTION=my-subscription,BACKLOG_THRESHOLD=1000

Step 2: Configure the Vigilmon HTTP Monitor

  1. Log in to Vigilmon → Add Monitor → HTTP.
  2. URL: your Cloud Function trigger URL.
  3. Check interval: 60 seconds.
  4. Response timeout: 10 seconds.
  5. Expected status: 200.
  6. JSON assertion: path status, expected value ok.
  7. Save and confirm the first check passes.

Step 3: Consumer Heartbeat for Pull Subscribers

If your service pulls messages from Pub/Sub, add a Vigilmon heartbeat ping after each successful processing batch. Vigilmon will alert you if the consumer stops sending pings.

Node.js pull consumer with heartbeat

// consumer.js
const { PubSub } = require("@google-cloud/pubsub");

const pubsub = new PubSub();
const subscription = pubsub.subscription(process.env.PUBSUB_SUBSCRIPTION);
let lastHeartbeat = 0;

subscription.on("message", async (message) => {
  try {
    await processMessage(message.data, message.attributes);
    message.ack();

    // Ping heartbeat at most once per minute
    const now = Date.now();
    if (now - lastHeartbeat > 60_000) {
      await fetch(process.env.VIGILMON_HEARTBEAT_URL, { method: "POST" });
      lastHeartbeat = now;
      console.log("Heartbeat sent");
    }
  } catch (err) {
    console.error("Message processing failed:", err);
    message.nack();
  }
});

subscription.on("error", (err) => {
  console.error("Subscription error:", err);
});

async function processMessage(data, attributes) {
  // Your business logic here
  const payload = JSON.parse(Buffer.from(data, "base64").toString());
  console.log("Processing:", payload);
}

Python pull consumer with heartbeat

# consumer.py
import os
import time
import requests
from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(
    os.environ["GOOGLE_CLOUD_PROJECT"],
    os.environ["PUBSUB_SUBSCRIPTION"],
)

last_heartbeat = 0

def callback(message):
    global last_heartbeat
    try:
        process_message(message.data, message.attributes)
        message.ack()

        now = time.time()
        if now - last_heartbeat > 60:
            requests.post(os.environ["VIGILMON_HEARTBEAT_URL"], timeout=5)
            last_heartbeat = now
            print("Heartbeat sent")
    except Exception as e:
        print(f"Processing failed: {e}")
        message.nack()

def process_message(data, attributes):
    import base64, json
    payload = json.loads(base64.b64decode(data).decode())
    print(f"Processing: {payload}")

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening on {subscription_path}")
try:
    streaming_pull_future.result()
except TimeoutError:
    streaming_pull_future.cancel()
    streaming_pull_future.result()

In Vigilmon, create a Heartbeat monitor with a grace period of 3–5 minutes for a high-volume subscription, or 10–15 minutes for a low-volume one.

Step 4: Dead-Letter Topic Monitoring

When a message fails to be delivered after the maximum delivery attempts, Pub/Sub forwards it to a dead-letter topic. Monitor this to detect processing failures early.

Configure a dead-letter topic

# Create the dead-letter topic
gcloud pubsub topics create my-subscription-dlq

# Update subscription to use dead-letter topic
gcloud pubsub subscriptions modify-push-config my-subscription \
  --dead-letter-topic=projects/my-project/topics/my-subscription-dlq \
  --max-delivery-attempts=5

Dead-letter subscriber with heartbeat alert

// dlq-monitor.js
const { PubSub } = require("@google-cloud/pubsub");

const pubsub = new PubSub();
const dlqSubscription = pubsub.subscription(process.env.PUBSUB_DLQ_SUBSCRIPTION);

dlqSubscription.on("message", async (message) => {
  // Any message here means a processing failure — withhold heartbeat
  console.error("Dead-letter message received:", {
    messageId: message.id,
    publishTime: message.publishTime,
    deliveryAttempt: message.deliveryAttempt,
    data: message.data.toString(),
  });

  // Acknowledge the DLQ message to stop redelivery
  message.ack();

  // Do NOT ping the heartbeat — Vigilmon will alert on missed heartbeat
});

dlqSubscription.on("error", (err) => {
  console.error("DLQ subscription error:", err);
});

// Ping DLQ heartbeat periodically when NO messages are arriving
setInterval(async () => {
  // This heartbeat is only sent when the interval fires without any DLQ messages
  // If DLQ messages are arriving, the interval callback will compete but the
  // message handler never pings, so the heartbeat eventually lapses
}, 60_000);

A cleaner approach: use a separate Cloud Functions timer that checks the DLQ subscription backlog via the Cloud Monitoring API and withholds the heartbeat if messages are present.

// dlq-check/index.js
const { http } = require("@google-cloud/functions-framework");
const { PubSub } = require("@google-cloud/pubsub");

const pubsub = new PubSub();

http("dlqCheck", async (req, res) => {
  const dlqSubscription = process.env.PUBSUB_DLQ_SUBSCRIPTION;
  const heartbeatUrl = process.env.VIGILMON_DLQ_HEARTBEAT_URL;

  try {
    // Pull up to 1 message with a short timeout to check if DLQ has content
    const subscription = pubsub.subscription(dlqSubscription);
    const [messages] = await subscription.pull({ maxMessages: 1 });

    if (messages.length > 0) {
      console.error(`DLQ has messages — withholding heartbeat`);
      // Nack so the message stays for inspection
      await subscription.nack(messages.map((m) => m.ackId));
      return res.status(500).json({ status: "dlq_has_messages" });
    }

    // DLQ empty — send heartbeat
    await fetch(heartbeatUrl, { method: "POST" });
    res.status(200).json({ status: "ok" });
  } catch (err) {
    console.error("DLQ check error:", err);
    res.status(500).json({ status: `error: ${err.message}` });
  }
});

Schedule this function every 5 minutes via Cloud Scheduler and create a separate Vigilmon heartbeat monitor for it.

Step 5: Push Subscription Monitoring

If you use push subscriptions (Pub/Sub delivers to your HTTP endpoint), add a health check to that endpoint and monitor it directly.

// push-handler/index.js
const express = require("express");
const app = express();

app.use(express.json());

// Pub/Sub push endpoint
app.post("/pubsub/push", async (req, res) => {
  try {
    const message = req.body.message;
    const data = Buffer.from(message.data, "base64").toString();
    await processMessage(JSON.parse(data), message.attributes);

    // Ping heartbeat on success
    await fetch(process.env.VIGILMON_HEARTBEAT_URL, { method: "POST" });
    res.status(204).send();
  } catch (err) {
    console.error("Push handler error:", err);
    res.status(500).json({ error: err.message });
  }
});

// Separate health endpoint for Vigilmon HTTP monitor
app.get("/health", (req, res) => {
  res.status(200).json({ status: "ok" });
});

Add both an HTTP monitor (for the /health endpoint) and a heartbeat monitor (pinged on each successful push delivery) in Vigilmon.

Step 6: Alerting Strategy

| Alert type | Recommended channel | |---|---| | HTTP health endpoint down | Email + PagerDuty webhook | | Consumer heartbeat missed | Slack webhook + email | | DLQ heartbeat missed | Immediate page — messages being dropped | | Backlog threshold exceeded | Alert via health endpoint returning 503 | | Status page | Public Vigilmon status page embed |

What Vigilmon Catches That Cloud Monitoring Misses

| Scenario | Cloud Monitoring | Vigilmon | |---|---|---| | Pull subscriber crashes | Needs oldest_unacked_message_age alarm | Consumer heartbeat missed — alert fires | | Dead-letter topic fills up | Needs explicit DLQ metric alert | DLQ heartbeat withheld — alert fires | | Push endpoint returns 500 | Needs log-based metric | HTTP monitor catches immediately | | Subscription deleted accidentally | Hard to alert reactively | HTTP health endpoint fails — alert fires | | GCP project quota exhausted | Your alarms may also break | External probe catches it |

Pub/Sub sits at the center of your event-driven architecture — when subscribers stop processing or messages get dead-lettered, downstream services starve silently. External monitoring from Vigilmon gives you the independent signal you can't get from inside your GCP project.

Start monitoring your Google Cloud Pub/Sub pipelines today — register free at vigilmon.online.

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