Streaming Billing Data with Pub/Sub

The BigQuery detailed export is authoritative but slow: rows land on an hours-scale, eventually-consistent cadence and keep restating for 30 days. When a budget threshold is crossed at 2 a.m. or a runaway job needs to be killed before the next billing cycle closes, a FinOps control loop cannot wait for the next export refresh. Cloud Billing solves this with a native budget notification topic in Cloud Pub/Sub — a push-based, sub-minute signal that fires the instant a budget’s spend crosses a configured threshold — and the same Pub/Sub primitives let you build a self-managed streaming layer that republishes granular cost deltas faster than the export’s batch load window. This page covers the streaming acquisition path: subscribing to billing signals over Pub/Sub with a durable, at-least-once-safe consumer, and it is the near-real-time complement to the batch sync covered in GCP BigQuery Billing Export Sync.

# Architecture Context & Data-Flow Position

Streaming ingestion sits in the same acquisition stage as the BigQuery export inside the four-stage model — acquisition, normalization, allocation, persistence — defined by the parent Cloud Billing Data Ingestion & Parsing reference, but it trades completeness for latency. Two distinct signal sources feed the topic layer. First, Cloud Billing’s built-in budget alerting publishes directly to a Pub/Sub topic you name when you create a budget — no polling, no export dependency, delivered within roughly a minute of the threshold crossing, but limited to coarse threshold-percentage events rather than line-item detail. Second, many teams layer a self-managed watcher (a Cloud Function or Dataflow streaming job) on top of the export table that polls for new export_time values and republishes each changed row as a granular cost-event message, giving downstream consumers sub-minute visibility into spend deltas without waiting for a full sync cycle. Both paths converge on the same consumer pattern: a durable pull subscription, flow-controlled streaming pull, and idempotent handling of at-least-once delivery.

The payload shapes differ by source but share a consumer contract:

A detailed comparison of when to reach for this streaming path instead of (or alongside) the BigQuery export lives in BigQuery Export vs Pub/Sub Streaming for Billing; the dedup strategy sketched below is expanded in Deduplicating Streamed Billing Events. Once a message is acknowledged, it hands off into the same asynchronous processing fabric described in Async Billing Data Processing Patterns.

# Core Implementation Patterns

# 1. Least-Privilege IAM for Publisher and Subscriber

Publisher and subscriber identities should never share a service account. The watcher job that republishes export deltas needs only roles/pubsub.publisher scoped to the finops-cost-events topic. The consumer needs only roles/pubsub.subscriber scoped to the specific finops-cost-events-sub subscription resource — not a project-wide grant, which would let it read every subscription in the project. The Pub/Sub service agent itself needs roles/pubsub.publisher on the dead-letter topic and roles/pubsub.subscriber on the source subscription; gcloud pubsub subscriptions create --dead-letter-topic grants these automatically, but Terraform-managed subscriptions must bind them explicitly.

# Consumer: subscription-scoped, not project-scoped.
gcloud pubsub subscriptions add-iam-policy-binding finops-cost-events-sub \
  --member="serviceAccount:[email protected]" \
  --role="roles/pubsub.subscriber"

# Publisher: topic-scoped only.
gcloud pubsub topics add-iam-policy-binding finops-cost-events \
  --member="serviceAccount:[email protected]" \
  --role="roles/pubsub.publisher"

# 2. Streaming Pull with Flow Control

subscribe() opens a long-lived bidirectional gRPC stream and hands each Message to a callback on a background thread pool. Without FlowControl, an unhealthy sink can cause the client library to keep pulling messages until memory is exhausted. Bound both the outstanding message count and byte volume:

from google.cloud import pubsub_v1

subscriber = pubsub_v1.SubscriberClient()
flow_control = pubsub_v1.types.FlowControl(
    max_messages=1000,          # cap outstanding, un-acked messages
    max_bytes=10 * 1024 * 1024,  # cap outstanding bytes (10 MB, one max-size message)
    max_lease_duration=600,      # never auto-extend leases past the 600s ack-deadline ceiling
)

# 3. Ordering Keys and Exactly-Once Dedup

The watcher publishes with ordering_key=row["project_id"] so a given project’s cost deltas are delivered to the subscriber in publish order — critical if a downstream consumer maintains a running per-project spend counter rather than an idempotent upsert. Ordering only holds within a single region and caps effective throughput on a hot key at roughly 1 MB/s, so keying on project.id (thousands of distinct keys) rather than billing_account.id (one key for the whole account) avoids a bottleneck. Subscription-level “exactly-once delivery” (enable_exactly_once_delivery=True) prevents redelivery to a second concurrent subscriber for the same lease, but it does not dedupe across process restarts, watcher republishes, or multiple subscriptions on the same topic — application-level content-hash dedup is still required for true idempotency.

subscription_path = subscriber.subscription_path("finops-prod", "finops-cost-events-sub")
streaming_pull_future = subscriber.subscribe(
    subscription_path,
    callback=handle_message,
    flow_control=flow_control,
)  # ordering is honored automatically when the subscription has enable_message_ordering=true

# 4. Dead-Letter Topic for Poison Messages

A malformed payload — a schema drift in the watcher’s JSON, a budget alert with an unexpected currency code — will fail the sink on every redelivery attempt forever unless it is capped. A dead-letter policy forwards the message to a separate topic after a bounded number of delivery attempts, so a single poison message cannot stall the whole subscription:

gcloud pubsub subscriptions create finops-cost-events-sub \
  --topic=finops-cost-events \
  --ack-deadline=120 \
  --enable-message-ordering \
  --dead-letter-topic=finops-cost-events-dlq \
  --max-delivery-attempts=5

# Production-Grade Python Streaming Subscriber Engine

The module below implements a durable streaming pull consumer for the billing signal topics: flow-controlled subscribe(), tenacity-retried sink writes for transient failures, content-hash deduplication to absorb at-least-once redelivery, and graceful shutdown on SIGTERM/SIGINT. Dependencies: google-cloud-pubsub>=2.21.0, tenacity>=8.2.0.

import hashlib
import json
import logging
import os
import signal
import time
from concurrent.futures import TimeoutError as FutureTimeoutError
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Callable, Dict, Optional

from google.api_core.exceptions import GoogleAPICallError, ServiceUnavailable
from google.cloud import pubsub_v1
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("gcp.billing.pubsub")

# Transient sink errors worth retrying before nack-ing the message.
TRANSIENT = (ServiceUnavailable, GoogleAPICallError, ConnectionError)


@dataclass(frozen=True)
class SubscriberConfig:
    """Identifiers and tuning knobs for one streaming pull consumer."""

    project_id: str
    subscription_id: str
    flow_max_messages: int = 1000
    flow_max_bytes: int = 10 * 1024 * 1024  # 10 MB — the hard per-message ceiling
    ack_deadline_seconds: int = 120         # must stay within the 10s–600s allowed range
    dedup_ttl_seconds: int = 3600           # how long a content hash is remembered

    @property
    def subscription_path(self) -> str:
        return f"projects/{self.project_id}/subscriptions/{self.subscription_id}"


@dataclass(frozen=True)
class BillingSignal:
    """Normalized view of a budget alert or streamed cost-event message."""

    message_id: str
    publish_time: datetime
    ordering_key: Optional[str]
    event_type: str            # "budget_alert" | "cost_event"
    project_id: Optional[str]
    display_name: str
    currency_code: str
    cost_amount: float
    budget_amount: Optional[float]
    threshold_pct: Optional[float]
    content_hash: str


class DedupCache:
    """In-process content-hash cache bounding redelivery-driven double processing.

    A single process's memory is NOT sufficient for a horizontally scaled
    subscriber fleet or a restarted pod — this is a demonstration cache.
    Production deployments back this with Redis or Firestore so the hash
    survives restarts and is visible across concurrent subscriber replicas;
    see the dedicated dedup deep dive for that design.
    """

    def __init__(self, ttl_seconds: int) -> None:
        self._ttl = ttl_seconds
        self._seen: Dict[str, float] = {}

    def seen(self, content_hash: str) -> bool:
        self._evict_expired()
        return content_hash in self._seen

    def remember(self, content_hash: str) -> None:
        self._seen[content_hash] = time.monotonic() + self._ttl

    def _evict_expired(self) -> None:
        now = time.monotonic()
        expired = [h for h, exp in self._seen.items() if exp <= now]
        for h in expired:
            del self._seen[h]


def compute_content_hash(payload: dict) -> str:
    """Stable sha256 over the fields that define event identity, not delivery.

    Sorting keys and excluding Pub/Sub's own message_id/publish_time means
    the SAME underlying cost event redelivered — or republished by a retried
    watcher run — hashes identically, which is what makes dedup possible.
    """
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def parse_message(message: pubsub_v1.subscriber.message.Message) -> BillingSignal:
    """Decode a raw Pub/Sub message into a normalized BillingSignal."""
    payload = json.loads(message.data.decode("utf-8"))
    event_type = message.attributes.get("event_type", "cost_event")
    content_hash = compute_content_hash(payload)
    return BillingSignal(
        message_id=message.message_id,
        publish_time=message.publish_time,
        ordering_key=message.ordering_key or None,
        event_type=event_type,
        project_id=payload.get("project_id"),
        display_name=payload.get("budgetDisplayName") or payload.get("display_name", ""),
        currency_code=payload.get("currencyCode", payload.get("currency_code", "USD")),
        cost_amount=float(payload.get("costAmount", payload.get("cost_amount", 0.0))),
        budget_amount=payload.get("budgetAmount"),
        threshold_pct=payload.get("alertThresholdExceeded"),
        content_hash=content_hash,
    )


class BillingStreamSubscriber:
    """Durable, flow-controlled consumer for GCP billing Pub/Sub signals."""

    def __init__(self, cfg: SubscriberConfig, sink: Callable[[BillingSignal], None]) -> None:
        self.cfg = cfg
        self.sink = sink
        self.client = pubsub_v1.SubscriberClient()
        self.dedup = DedupCache(cfg.dedup_ttl_seconds)

    @retry(
        retry=retry_if_exception_type(TRANSIENT),
        stop=stop_after_attempt(4),
        wait=wait_exponential_jitter(initial=1, max=20),
        reraise=True,
    )
    def _write_sink(self, signal: "BillingSignal") -> None:
        self.sink(signal)

    def handle_message(self, message: pubsub_v1.subscriber.message.Message) -> None:
        try:
            signal = parse_message(message)
        except (json.JSONDecodeError, KeyError, ValueError) as exc:
            # Permanent, non-transient failure — ack so it does not clog the
            # subscription; the payload is unrecoverable, not merely delayed.
            logger.error("unparseable message %s: %s", message.message_id, exc)
            message.ack()
            return

        if self.dedup.seen(signal.content_hash):
            logger.info(
                "duplicate suppressed: %s (hash=%s)", signal.message_id, signal.content_hash[:12]
            )
            message.ack()
            return

        try:
            self._write_sink(signal)
        except TRANSIENT as exc:
            logger.warning("transient sink failure on %s: %s; nacking for redelivery",
                            signal.message_id, exc)
            message.nack()
            return
        except Exception:
            logger.exception("permanent sink failure on %s; nacking toward dead-letter",
                              signal.message_id)
            message.nack()
            return

        self.dedup.remember(signal.content_hash)
        message.ack()
        logger.info("processed %s project=%s cost=%.2f %s",
                    signal.event_type, signal.project_id, signal.cost_amount, signal.currency_code)

    def run(self) -> None:
        flow_control = pubsub_v1.types.FlowControl(
            max_messages=self.cfg.flow_max_messages,
            max_bytes=self.cfg.flow_max_bytes,
            max_lease_duration=self.cfg.ack_deadline_seconds,
        )
        future = self.client.subscribe(
            self.cfg.subscription_path,
            callback=self.handle_message,
            flow_control=flow_control,
        )
        logger.info("listening on %s (ack_deadline=%ds, max_messages=%d)",
                    self.cfg.subscription_path, self.cfg.ack_deadline_seconds,
                    self.cfg.flow_max_messages)

        def _shutdown(signum, _frame) -> None:
            logger.info("received signal %d; cancelling streaming pull", signum)
            future.cancel()

        signal.signal(signal.SIGTERM, _shutdown)
        signal.signal(signal.SIGINT, _shutdown)

        with self.client:
            try:
                future.result()
            except FutureTimeoutError:
                future.cancel()
                future.result()


def bigquery_sink(signal: BillingSignal) -> None:
    """Placeholder idempotent sink — swap for a streaming insert or upsert."""
    logger.debug("sink write: %s", signal)


def main() -> None:
    cfg = SubscriberConfig(
        project_id=os.environ.get("GCP_PROJECT", "finops-prod"),
        subscription_id=os.environ.get("PUBSUB_SUBSCRIPTION", "finops-cost-events-sub"),
    )
    subscriber = BillingStreamSubscriber(cfg, sink=bigquery_sink)
    subscriber.run()


if __name__ == "__main__":
    main()

# Schema Reference Table

Both the budget notification payload and the watcher-published cost-event payload arrive as JSON in message.data; this table maps the raw fields onto the normalized BillingSignal model the engine above produces.

# Operational Considerations

• Ack deadline: 10 seconds default, 600 seconds (10 minutes) maximum. Set it high enough to cover your slowest sink write plus retry backoff, or rely on the client library’s automatic lease extension — but the extension itself is capped at max_lease_duration, which cannot exceed 600s. A sink write that can legitimately take longer must ack early and continue work asynchronously.
• Message size ceiling: 10 MB. Both message.data and attributes count toward the limit. Billing signal payloads are small (well under 1 KB), so this rarely binds directly, but it does cap flow_control.max_bytes sizing — one worst-case message still fits inside the 10 MB flow-control budget.
• Retention: 7 days default for unacknowledged messages, configurable 10 minutes to 31 days via messageRetentionDuration. Acknowledged messages are purged immediately unless retain_acked_messages=True is set for replay/audit purposes.
• Delivery is at-least-once, not exactly-once, by default. Redelivery after a nack, a lease expiry, or a subscriber crash between sink write and ack is normal and expected — this is why compute_content_hash and DedupCache exist in the engine above, not as a defensive afterthought.
• Ordering key throughput ceiling: roughly 1 MB/s per key. Keying on project.id spreads load across thousands of keys; keying on a single billing_account.id collapses every project’s events onto one ordered queue and throttles the whole pipeline to that ceiling.
• Dead-letter delivery attempts: configurable 5–100. Five is a reasonable default for billing signals — enough to survive a transient sink outage, low enough that a genuinely malformed payload reaches the dead-letter topic within seconds rather than hours of futile redelivery.
• Flow control has no default cap. Without an explicit FlowControl, the client pulls as fast as the network allows; an unbounded subscriber against a slow sink is the most common cause of an out-of-memory subscriber pod. Always set max_messages and max_bytes explicitly, as shown above.
• Monitoring hooks. Export subscription/num_undelivered_messages, subscription/oldest_unacked_message_age, and subscription/dead_letter_message_count from Cloud Monitoring; alert when the oldest unacked age exceeds a few multiples of the ack deadline, which signals the sink has stalled rather than a quiet spend day.

# Troubleshooting

Duplicate cost events double-count spend downstream. Root cause: at-least-once redelivery after a sink write succeeded but the ack was lost (pod restart, network blip) before Pub/Sub recorded it. Detection: the same content_hash appears in sink writes more than once, correlated with subscriber restarts in the logs. Remediation: persist the dedup cache outside process memory (Redis or Firestore, TTL-bounded) so it survives restarts and is shared across replicas — see Deduplicating Streamed Billing Events for the persisted-cache design.

Backlog grows without bound. Root cause: the sink (a BigQuery streaming insert, a downstream API) is slower than the flow-controlled pull rate, or an undersized ack_deadline_seconds causes messages to expire and redeliver before the sink finishes, compounding the backlog. Detection: subscription/num_undelivered_messages and subscription/oldest_unacked_message_age climb monotonically. Remediation: raise ack_deadline_seconds toward the 600s ceiling, lower flow_control.max_messages to reduce concurrent in-flight work, or run additional subscriber replicas against the same subscription — Pub/Sub load-balances streaming pull across all attached clients automatically.

Messages land in the dead-letter topic unexpectedly. Root cause: the sink raises a non-transient exception (a schema-violating payload, a permanently invalid currency code) on every delivery attempt, so the message exhausts max_delivery_attempts and diverts. Detection: finops-cost-events-dlq has a nonzero message count; the message carries a CloudPubSubDeadLetterSourceSubscription attribute. Remediation: distinguish permanent parse failures (ack immediately, log, never nack) from transient sink failures (nack, let it redeliver) as the handle_message method above does — nacking a permanently broken payload only delays its arrival in the dead-letter topic.

One project’s events stop advancing while others keep flowing. Root cause: message ordering is enabled, and a single nacked message for a given ordering_key blocks all subsequent messages under that same key until it is acknowledged. Detection: oldest_unacked_message_age is elevated only for messages carrying one specific ordering_key. Remediation: keep ordering-key handlers fast and side-effect-light; route unrecoverable payloads to ack-plus-manual-dead-letter-publish rather than repeated nack so the head-of-line block clears.

PermissionDenied on subscribe(). Root cause: the service account has roles/pubsub.subscriber at the project level but the workload identity binding for the running pod points at a different account, or the role was granted on the topic instead of the subscription. Detection: the client raises google.api_core.exceptions.PermissionDenied immediately on stream open. Remediation: grant roles/pubsub.subscriber directly on the subscription resource and confirm the GKE/Cloud Run workload identity binding matches the account the grant was issued to.

# Frequently Asked Questions

# Related

• Cloud Billing Data Ingestion & Parsing — the parent reference defining the four-stage acquisition→normalization→allocation→persistence model this streaming path implements.
• BigQuery Export vs Pub/Sub Streaming for Billing — a head-to-head comparison of latency, completeness, and cost between the two acquisition paths.
• Deduplicating Streamed Billing Events — the persisted, cross-replica dedup design that hardens the in-process cache shown here.
• GCP BigQuery Billing Export Sync — the batch acquisition path this streaming layer complements rather than replaces.
• Async Billing Data Processing Patterns — the broader asynchronous processing fabric that acknowledged messages hand off into.

Up: Cloud Billing Data Ingestion & Parsing