A distributed event-driven pipeline fails differently from a monolith. When a single service goes down, you get an error page. When an event-driven pipeline degrades, it's often invisible at first — messages pile up in a subscription backlog, a DAG silently misses its SLA, a dead-letter queue grows without triggering any alert. By the time a user notices, the system has been degraded for hours.
Observability in an event-driven system isn't optional, and it isn't just application logging. It's knowing the health of every layer — the broker, the consumers, the task queues, and the orchestration layer — before users are affected. This final post covers how PulseCart monitors each layer, what alerts matter, and what actually breaks at 10x current scale.
Layer 1: Pub/Sub Subscription Backlog
The most important metric in a Pub/Sub-based pipeline is subscription/num_undelivered_messages — the number of messages waiting to be delivered on a subscription. A healthy consumer processes messages as fast as they arrive. A growing backlog means either the consumer is too slow, crashing, or not running at all.
Setting Up a Backlog Alert
# monitoring/alerts.py — create alerts programmatically via Cloud Monitoring API
from google.cloud import monitoring_v3
from google.protobuf.duration_pb2 import Duration
client = monitoring_v3.AlertPolicyServiceClient()
project_name = f"projects/your-gcp-project-id"
backlog_alert = monitoring_v3.AlertPolicy(
display_name="PulseCart: Commerce Events Backlog High",
conditions=[
monitoring_v3.AlertPolicy.Condition(
display_name="Undelivered messages > 1000",
condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold(
filter='resource.type="pubsub_subscription" '
'AND resource.labels.subscription_id="sub-realtime-consumer" '
'AND metric.type="pubsub.googleapis.com/subscription/num_undelivered_messages"',
comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
threshold_value=1000,
duration=Duration(seconds=300), # sustained for 5 minutes before alerting
aggregations=[
monitoring_v3.Aggregation(
alignment_period=Duration(seconds=60),
per_series_aligner=monitoring_v3.Aggregation.Aligner.ALIGN_MAX,
)
],
),
)
],
notification_channels=["projects/your-gcp-project-id/notificationChannels/YOUR_CHANNEL_ID"],
alert_strategy=monitoring_v3.AlertPolicy.AlertStrategy(
auto_close=Duration(seconds=1800)
),
)
client.create_alert_policy(name=project_name, alert_policy=backlog_alert)
The 5-minute duration window prevents alert noise from brief spikes — a surge of order.placed events during a flash sale will temporarily spike the backlog, but if the consumer is healthy it clears within minutes. You want alerts for sustained backlogs, not momentary ones.
Track these metrics per subscription, not per topic:
| Metric |
Alert Threshold |
What It Means |
num_undelivered_messages |
> 1000 for 5 min |
Consumer falling behind |
oldest_unacked_message_age |
> 10 min |
Messages stuck — consumer may be crashing |
num_undelivered_messages on dead-letter |
> 0 |
Failed messages accumulating |
oldest_unacked_message_age is often more useful than raw backlog size. A backlog of 5000 messages being processed quickly is fine. A backlog of 50 messages where the oldest is 30 minutes old means something is stuck.
Layer 2: Cloud Run Consumer Health
Cloud Run surfaces two categories of metrics that matter for PulseCart's consumers:
Request metrics — latency, error rate, request count. These tell you how the consumer is performing per invocation.
Instance metrics — active instances, startup latency. These tell you whether autoscaling is keeping up with load.
# Cloud Monitoring dashboard config (as code via Terraform)
# modules/monitoring/main.tf
resource "google_monitoring_dashboard" "pulsecart" {
dashboard_json = jsonencode({
displayName = "PulseCart Pipeline Health"
gridLayout = {
columns = 2
widgets = [
{
title = "Consumer Error Rate"
xyChart = {
dataSets = [{
timeSeriesQuery = {
timeSeriesFilter = {
filter = join(" AND ", [
"resource.type=\"cloud_run_revision\"",
"resource.labels.service_name=\"pulsecart-consumer\"",
"metric.type=\"run.googleapis.com/request_count\"",
"metric.labels.response_code_class!=\"2xx\""
])
}
}
}]
}
},
{
title = "Consumer P99 Latency"
xyChart = {
dataSets = [{
timeSeriesQuery = {
timeSeriesFilter = {
filter = join(" AND ", [
"resource.type=\"cloud_run_revision\"",
"resource.labels.service_name=\"pulsecart-consumer\"",
"metric.type=\"run.googleapis.com/request_latencies\""
])
}
}
}]
}
}
]
}
})
}
Alert thresholds for Cloud Run:
| Metric |
Alert Threshold |
Action |
| Error rate |
> 5% for 3 min |
Page on-call, check consumer logs |
| P99 latency |
> 25s (of 30s ack deadline) |
Consumer processing too slowly |
| Max instances reached |
Sustained at ceiling |
Scale up max_instance_count |
P99 latency approaching the Pub/Sub ack deadline is a leading indicator of trouble. If the consumer takes 28 seconds on a 30-second deadline, any variance causes missed acks and redelivery — which compounds the backlog problem.
Layer 3: Cloud Tasks Queue Depth
Cloud Tasks exposes queue-level metrics that tell you whether delayed work is executing on schedule or backing up.
# Cloud Monitoring alert for Cloud Tasks — in Terraform alerting config
filter: >
resource.type="cloudtasks.googleapis.com/Queue"
AND resource.labels.queue_id="pulsecart-cart-reminders"
AND metric.type="cloudtasks.googleapis.com/queue/depth"
threshold_value: 10000
duration: 600s # 10 minutes
For PulseCart's cart reminder queue, a depth of 10,000 tasks sustained for 10 minutes means either the handler is failing (Cloud Tasks retrying everything) or there was a genuine spike in cart abandonments. Check the task dispatch error rate alongside queue depth to distinguish between the two.
Key Cloud Tasks metrics:
| Metric |
What to Watch |
queue/depth |
Total tasks waiting — alert if sustained high |
queue/task_attempt_failures |
Tasks failing on dispatch — indicates handler errors |
queue/task_attempt_count |
Retry rate — high retries signal handler instability |
Layer 4: Airflow DAG Monitoring
Cloud Composer exposes Airflow metrics via Cloud Monitoring. The ones that matter for PulseCart:
# dags/monitoring_check.py — a lightweight DAG that validates upstream DAGs ran successfully
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import DagRun
from airflow.utils.state import State
from datetime import datetime, timedelta
import pendulum
import logging
logger = logging.getLogger(__name__)
def check_critical_dags(**context):
critical_dags = [
"pulsecart_nightly_aggregation",
"pulsecart_dead_letter_reconciliation",
]
execution_date = context["data_interval_start"]
lookback = execution_date - timedelta(hours=25) # give 1-hour buffer
failed = []
for dag_id in critical_dags:
runs = DagRun.find(
dag_id=dag_id,
execution_start_date=lookback,
execution_end_date=execution_date,
)
if not runs:
failed.append(f"{dag_id}: no run found in last 25 hours")
elif runs[-1].state != State.SUCCESS:
failed.append(f"{dag_id}: last run state={runs[-1].state}")
if failed:
raise ValueError(f"Critical DAG failures detected:\n" + "\n".join(failed))
logger.info("All critical DAGs ran successfully.")
with DAG(
dag_id="pulsecart_dag_health_check",
schedule_interval="0 6 * * *", # runs at 06:00 UTC, after nightly DAGs complete
start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
catchup=False,
tags=["pulsecart", "monitoring"],
) as dag:
PythonOperator(
task_id="check_critical_dags",
python_callable=check_critical_dags,
)
This DAG runs at 06:00 UTC — after the nightly aggregation and metrics DAGs should have completed — and raises if either didn't succeed. Airflow's built-in alerting fires on DAG failure, which triggers the notification channel.
Beyond this, configure SLA misses directly on your critical DAGs:
# Add to nightly_aggregation DAG definition
with DAG(
dag_id="pulsecart_nightly_aggregation",
sla_miss_callback=notify_sla_miss, # custom callback that pages on-call
...
) as dag:
Layer 5: Structured Logging Across the Pipeline
Individual metric alerts tell you something is wrong. Structured logs tell you why. Every service in PulseCart logs in JSON with consistent fields so Cloud Logging can correlate events across the pipeline by event_id.
# services/logging.py
import logging
import json
import sys
class StructuredLogger:
def __init__(self, service_name: str):
self.service_name = service_name
self.logger = logging.getLogger(service_name)
def info(self, message: str, **kwargs):
self._log("INFO", message, **kwargs)
def error(self, message: str, **kwargs):
self._log("ERROR", message, **kwargs)
def warning(self, message: str, **kwargs):
self._log("WARNING", message, **kwargs)
def _log(self, severity: str, message: str, **kwargs):
entry = {
"severity": severity,
"message": message,
"service": self.service_name,
**kwargs
}
print(json.dumps(entry), file=sys.stdout)
# Usage in consumer
logger = StructuredLogger("pulsecart-consumer")
logger.info(
"Event processed",
event_id="evt_a3f9b21c4d8e",
event_type="order.placed",
user_id="usr_8821",
latency_ms=42,
)
With event_id on every log entry, a Cloud Logging query like jsonPayload.event_id="evt_a3f9b21c4d8e" traces a single event from ingestion through the producer, across Pub/Sub, into the consumer, and through any downstream Cloud Tasks — across services, without distributed tracing infrastructure.
What Breaks at 10x Scale
PulseCart currently processes roughly 1 million events per day across all topics. At 10x — 10 million events — here's what changes:
Pub/Sub ordering keys become a bottleneck. Ordering keys force sequential delivery per key value within a subscription. At 10x user volume, popular users (or bots) generate enough events to create ordering-key hot spots that slow delivery for the entire subscription. The fix is to shard high-volume subscriptions or relax ordering requirements on non-critical event types.
The Redis idempotency store needs sharding. A single Redis instance has throughput limits. At 10x event volume, the idempotency check (GET processed:{event_id}) becomes a bottleneck. Cloud Memorystore for Redis Cluster (available in GCP) handles this via automatic sharding, or you can partition the key space across multiple Redis instances at the application layer.
Cloud SQL connections exhaust. Cloud Run scales to many instances under load, each holding a connection pool. At 10x scale, the total connection count across all consumer instances can exceed PostgreSQL's max_connections. The fix is PgBouncer as a connection pooler in front of Cloud SQL, or Cloud SQL's built-in connection pooling via the Auth Proxy with a --max-connections flag.
Airflow worker concurrency caps out. The dead-letter reconciliation DAG pulls up to 500 messages per run. At 10x dead-letter volume, a single worker takes too long per topic. The fix is dynamic task mapping — Airflow 2.3+ lets you generate task instances at runtime, so each dead-letter topic gets its own worker in parallel rather than sequential processing.
Cloud Composer cost grows non-linearly. At 10x DAG runs with more complex dependencies, Composer's GKE cluster costs scale with you. At that point, re-evaluate whether managed Airflow is the right fit or whether a lighter orchestrator (Prefect, Dagster, or a self-hosted Airflow on GKE Autopilot) makes more economic sense.
The Observability Stack, Summarised
| Layer |
Primary Metric |
Alert Condition |
Tool |
| Pub/Sub |
oldest_unacked_message_age |
> 10 minutes |
Cloud Monitoring |
| Cloud Run |
Error rate, P99 latency |
> 5% errors, P99 > 25s |
Cloud Monitoring |
| Cloud Tasks |
Queue depth, failure rate |
Depth > 10k sustained |
Cloud Monitoring |
| Airflow |
DAG run state, SLA miss |
Any critical DAG failure |
Airflow + Cloud Monitoring |
| Cross-service |
event_id trace |
N/A — for debugging |
Cloud Logging |
Closing Thoughts
This series set out to build PulseCart the way a production system actually gets built — not a hello-world demo, but a full pipeline with real tradeoffs at every layer. Here's what we covered across the eight posts:
Day 0 — What PulseCart is and what to expect
Day 1 — Why request-response breaks at scale and how to design an event taxonomy
Day 2 — Pub/Sub topics, subscriptions, ordering keys, and dead-letter topics
Day 3 — FastAPI producer with Pydantic validation and at-least-once delivery semantics
Day 4 — Cloud Run consumers, Redis idempotency, and Cloud Tasks for delayed workflows
Day 5 — Airflow DAGs for batch aggregation, dead-letter reconciliation, and metrics
Day 6 — Full Terraform infrastructure and zero-downtime GitHub Actions CI/CD
Day 7 — Observability, alerting, and what breaks at 10x scale
The GCP-native stack — Pub/Sub, Cloud Tasks, Cloud Run, Cloud Composer — won't be the right fit for every team. If you're already running Kafka, the producer/consumer patterns from Days 3 and 4 translate directly. If you're on AWS, SNS/SQS maps closely to what Pub/Sub does here. The infrastructure changes; the architecture doesn't.
Thanks for following along. If you found something useful, wrong, or worth pushing back on — the comments are open.
