BigQuery Export vs Pub/Sub Streaming for Billing

Teams building the Streaming Billing Data with Pub/Sub pipeline routinely conflate two GCP mechanisms that both get called “real-time billing” but solve opposite problems. The BigQuery billing export is the authoritative, queryable ledger — full line-item detail (SKU, project, labels, credits) refreshed several times a day, but with no push mechanism and a settlement window that stretches to roughly 30 days before figures are final. Pub/Sub budget notifications are the opposite: a push message that lands in seconds once a configured spend threshold is crossed, but the payload is a single aggregate number per budget, never a line item. Building an alerting system on the notification alone gives you speed with no detail; querying the export on a timer gives you detail with no speed. Production billing-signal systems need both — the export as source of truth, Pub/Sub as the trigger that tells you when to go look.

# Root Cause & Failure Modes

The failure mode is almost always an architectural shortcut taken because the two mechanisms look similar from a distance — both are “GCP tells me about spend.” In practice they diverge on every axis that matters for automation:

• Treating the notification payload as a cost feed. A budget notification’s JSON body carries costAmount, budgetAmount, currencyCode, and which threshold rule tripped — nothing else. There is no project.id, no sku.id, no resource-level breakdown. Teams that wire Slack or PagerDuty alerts straight off this payload can say “spend crossed $50k” but cannot answer “which service” without a follow-up query. Google also only publishes a message when Cloud Billing’s own cost data refreshes — a handful of times a day, not continuously — so a budget can silently sit above threshold for hours between notification pushes if the eligible rule already fired.
• Polling the BigQuery export as if it were a live feed. The export refreshes several times per day, but the rows are not finalized: committed-use discount reallocations, credits, and tax corrections keep rewriting usage_start_time days for up to the same ~30-day restatement window described in Incremental Sync Strategies for GCP Billing Exports. A naive cron job that queries the full table every few minutes to “catch spikes early” burns query bytes for no latency benefit — the underlying data hasn’t moved since the last refresh — and on-demand BigQuery pricing bills roughly $6.25 per TiB scanned, so a tight poll loop against a large export is an expensive way to detect nothing.
• Ignoring the threshold-rule limit. Each GCP budget supports at most five threshold rules (e.g., 50%, 90%, 100%, 110%, 150% of the budget amount). Teams that need finer-grained escalation stages discover this cap only after trying to add a sixth rule, and end up either creating parallel budgets or accepting coarse-grained triggers.

The fix is not choosing one mechanism over the other — it’s using each for what it is actually good at: Pub/Sub as a cheap, event-driven trip wire, and the BigQuery export as the queryable ledger you reconcile against every time the trip wire fires.

# Production Pipeline Architecture

The reconciliation engine below runs a four-phase model that treats the notification strictly as a trigger and the export as the fact:

1. Trigger. A Pub/Sub push or pull subscription delivers a budget notification the instant a threshold rule crosses. This is the only part of the system that is genuinely event-driven.
2. Deduplicate. Pub/Sub delivers at-least-once, so the same threshold crossing can arrive twice. The orchestrator keys each signal on (billing_account_id, budget_display_name, cost_interval_start, threshold_percent) before doing any work — the same natural-key discipline covered in depth in Deduplicating Streamed Billing Events.
3. Reconcile. For every de-duplicated signal, query the BigQuery export — the same curated table maintained by GCP BigQuery Billing Export Sync — for the actual net cost (cost plus summed credits) over the notified interval.
4. Escalate or acknowledge. Only after the reconciliation query succeeds does the orchestrator acknowledge the Pub/Sub message. If notified and queried cost diverge by more than a small tolerance, that drift itself is the signal worth escalating, because it usually means the notification fired against partial-day data that the export hasn’t finished settling.

This is deliberately not a replacement for the export sync — it is a thin, event-driven layer that decides when to spend a BigQuery query, rather than polling on a fixed schedule.

# Step-by-Step Python Implementation

The module below pulls pending budget notifications, deduplicates them, reconciles each against the BigQuery export, and only acknowledges a message after its reconciliation query commits successfully — so a crash mid-run redelivers the signal instead of losing it.

import os
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone

from google.cloud import bigquery, pubsub_v1
from google.api_core.exceptions import GoogleAPIError
from google.api_core.retry import Retry, if_transient_error

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

SUBSCRIPTION_ID = os.getenv("BUDGET_SUBSCRIPTION_ID", "billing-budget-notifications-sub")
BQ_DATASET = "finops_billing"
BQ_TABLE = "gcp_billing_incremental"  # maintained by the export-sync engine
DRIFT_TOLERANCE_PCT = 2.0  # acceptable variance between notified and queried net cost

bq_retry = Retry(initial=1.0, maximum=30.0, multiplier=2.0,
                 predicate=if_transient_error, timeout=120.0)


@dataclass(frozen=True)
class BudgetSignal:
    """A decoded Pub/Sub budget notification -- a trigger, not a line item."""
    billing_account_id: str
    budget_display_name: str
    cost_amount: float
    currency_code: str
    threshold_percent: float
    cost_interval_start: datetime

    @property
    def dedup_key(self) -> str:
        # One signal per (budget, interval, threshold) -- Pub/Sub delivers at-least-once.
        return (f"{self.billing_account_id}:{self.budget_display_name}:"
                f"{self.cost_interval_start.isoformat()}:{self.threshold_percent}")


@dataclass
class ReconciliationResult:
    signal: BudgetSignal
    queried_cost: float
    drift_pct: float
    reconciled: bool


class BillingSignalOrchestrator:
    """Pub/Sub supplies the trigger; the BigQuery export supplies the truth."""

    def __init__(self, project_id: str):
        self.project_id = project_id
        self.subscriber = pubsub_v1.SubscriberClient()
        self.bq_client = bigquery.Client(project=project_id)
        self.subscription_path = self.subscriber.subscription_path(project_id, SUBSCRIPTION_ID)
        self._seen: set[str] = set()  # per-run dedup; back with Firestore/Redis across runs

    def pull_budget_signals(self, max_messages: int = 20) -> list[tuple[BudgetSignal, str]]:
        """Pull pending notifications and return (signal, ack_id) pairs, deduplicated."""
        response = self.subscriber.pull(
            request={"subscription": self.subscription_path, "max_messages": max_messages},
            retry=bq_retry,
        )
        results = []
        for received in response.received_messages:
            try:
                payload = json.loads(received.message.data.decode("utf-8"))
                signal = BudgetSignal(
                    billing_account_id=payload["billingAccountId"],
                    budget_display_name=payload["budgetDisplayName"],
                    cost_amount=float(payload["costAmount"]),
                    currency_code=payload.get("currencyCode", "USD"),
                    threshold_percent=float(payload.get("alertThresholdExceeded", 0.0)),
                    cost_interval_start=datetime.fromisoformat(
                        payload["costIntervalStart"]).replace(tzinfo=timezone.utc),
                )
            except (json.JSONDecodeError, KeyError, ValueError) as exc:
                logger.error("Malformed budget notification, skipping: %s", exc)
                continue
            if signal.dedup_key in self._seen:
                logger.info("Duplicate delivery for %s, will ack without reprocessing.",
                            signal.dedup_key)
            else:
                self._seen.add(signal.dedup_key)
            results.append((signal, received.ack_id))
        return results

    def reconcile_against_export(self, signal: BudgetSignal) -> ReconciliationResult:
        """Query the export -- the authoritative source -- for the notified window
        and compare. The notification claims a number; this confirms it.
        """
        query = f"""
        SELECT SUM(cost) + SUM((SELECT IFNULL(SUM(c.amount), 0) FROM UNNEST(credits) c)) AS net_cost
        FROM `{self.project_id}.{BQ_DATASET}.{BQ_TABLE}`
        WHERE billing_account_id = @billing_account_id
          AND usage_start_time >= @interval_start
        """
        job_config = bigquery.QueryJobConfig(query_parameters=[
            bigquery.ScalarQueryParameter("billing_account_id", "STRING", signal.billing_account_id),
            bigquery.ScalarQueryParameter("interval_start", "TIMESTAMP", signal.cost_interval_start),
        ])
        job = self.bq_client.query(query, job_config=job_config, retry=bq_retry)
        row = next(iter(job.result()))
        queried_cost = float(row.net_cost or 0.0)
        drift_pct = abs(queried_cost - signal.cost_amount) / queried_cost * 100 if queried_cost else 0.0
        reconciled = drift_pct <= DRIFT_TOLERANCE_PCT
        if not reconciled:
            logger.warning("Drift %.2f%% for %s: notified=%.2f queried=%.2f",
                            drift_pct, signal.budget_display_name, signal.cost_amount, queried_cost)
        return ReconciliationResult(signal, queried_cost, drift_pct, reconciled)

    def acknowledge(self, ack_ids: list[str]) -> None:
        if ack_ids:
            self.subscriber.acknowledge(
                request={"subscription": self.subscription_path, "ack_ids": ack_ids})

    def run_once(self) -> list[ReconciliationResult]:
        results, ack_ids = [], []
        for signal, ack_id in self.pull_budget_signals():
            try:
                results.append(self.reconcile_against_export(signal))
                ack_ids.append(ack_id)  # only ack after a successful reconciliation query
            except GoogleAPIError as exc:
                logger.error("Reconciliation failed for %s, leaving unacked: %s",
                             signal.dedup_key, exc)
        self.acknowledge(ack_ids)
        return results


def main() -> None:
    orchestrator = BillingSignalOrchestrator(project_id=os.environ["GCP_PROJECT_ID"])
    for result in orchestrator.run_once():
        status = "OK" if result.reconciled else "DRIFT"
        logger.info("[%s] %s: notified=%.2f queried=%.2f (%.2f%% drift)",
                    status, result.signal.budget_display_name,
                    result.signal.cost_amount, result.queried_cost, result.drift_pct)


if __name__ == "__main__":
    main()

# Verification & Testing

• Replay a known crossing. Take a budget notification you already have logs for, feed its payload through reconcile_against_export, and confirm the drift falls under DRIFT_TOLERANCE_PCT. A miss usually means the export hadn’t refreshed for that interval yet — check the export’s last refresh timestamp before assuming a bug.
• Redelivery idempotency. Push the same notification payload through pull_budget_signals twice in one run. The second occurrence must be flagged as a duplicate via dedup_key and must not trigger a second reconciliation query — assert on len(self._seen) staying at one.
• Ack-after-success ordering. Force reconcile_against_export to raise mid-run (mock a GoogleAPIError) and confirm the corresponding ack_id is never passed to acknowledge. That message must be redelivered on the next pull, not silently dropped.
• Dry-run the reconciliation query. Run it with QueryJobConfig(dry_run=True) before deploying and confirm total_bytes_processed stays proportional to one billing account’s slice, not the whole export — a missing WHERE predicate here turns a cheap trigger-driven check into a full-table scan on every threshold crossing.

# Common Pitfalls Checklist

• Trusting costAmount as the final number. It reflects whatever cost data was available when the notification fired, not the settled figure — always reconcile against the export before acting on it.
• Skipping deduplication. Pub/Sub’s at-least-once delivery means a single threshold crossing can trigger two reconciliation queries and two alerts — key on the natural signal identity as shown above.
• Acking before reconciliation succeeds. Ack early and a failed reconciliation is gone forever; ack only after the BigQuery query commits.
• Polling the export on a fixed timer “for speed.” The export’s own refresh cadence bounds how often the data changes — a tighter poll just burns bytes scanned without buying latency. Let the Pub/Sub trigger decide when to query.
• Comparing gross cost instead of net cost. The export’s cost column excludes credits — sum the credits array into the comparison, as the reconciliation query does, or drift will look larger than it is.

# Frequently Asked Questions

# Related

• Streaming Billing Data with Pub/Sub — the parent pipeline this reconciliation pattern plugs into.
• Deduplicating Streamed Billing Events — the natural-key dedup discipline behind the dedup_key logic used here.
• GCP BigQuery Billing Export Sync — the engine that maintains the curated export table this module queries as source of truth.
• Incremental Sync Strategies for GCP Billing Exports — why the export itself lags and restates for up to 30 days, which bounds how tight the drift tolerance above can safely be.
• Cloud Billing Data Ingestion & Parsing — the broader acquisition model these two GCP signal sources both feed.

Up: Streaming Billing Data with Pub/Sub · Home: Cloud Cost Optimization & FinOps Automation