# Observability and Scaling: Monitoring Event Lag and Handling Failure 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 ```python # 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. ```python # 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. ```yaml # 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: ```python # 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: ```python # 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`. ```python # 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.