"""Production-grade cost-analysis service for the cloud-cost dashboard. Three concerns live here: 1. **Ingestion** — accept ``CostFact`` rows from any cloud driver and upsert them into ``cmdb_cost_facts``. Idempotent via the natural- key sha256 (``fact_key``); concurrent ingest runs converge. 2. **Allocation** — two-stage attribution: a. Walk the priority-ordered ``cmdb_cost_allocation_rules`` ladder first (admin-authored overrides, e.g. "RG=foundry-rg → split off the Azure OpenAI subscription onto applicant X"). b. When no rule matches, fall back to the fact's ``cmdb_cloud_scopes.owner_person_id`` — the default subscription / project / sub-account owner the fact lives under. This is what makes "no rule, all default attribution" work without requiring the admin to author one rule per scope. The chosen path is recorded in ``cmdb_cost_facts.auto_owner_source`` so the dashboard can label rule-driven vs scope-owner-fallback vs genuinely unattributed facts. Recomputable in-place via :func:`recompute_allocation` after a rule edit OR a scope-ownership change; the dashboard's "重算分摊" button drives this. 3. **Analytics** — multi-dim breakdown, top-resources, time series, month-end forecast, and anomaly detection. All read directly from ``cmdb_cost_facts`` with composite indexes; no MV / cache layer (TiDB handles the working-set queries within ~200ms at the current scale and the cost of MV maintenance would buy us nothing). Constraints honoured: * single TiDB cluster (no Redshift / ClickHouse / dedicated OLAP); * no ML / numpy dependency for forecast / anomaly — pure stdlib statistics, since the volumes (one-org cost data) trivially fit in process memory; * idempotent re-runs (cron or admin) — same input window converges to the same DB state without duplication or drift. """ from __future__ import annotations import asyncio import hashlib import json import math import re import statistics import time import uuid from collections import defaultdict from collections.abc import Awaitable, Callable, Sequence from dataclasses import replace as _dc_replace from datetime import UTC, date, datetime, timedelta from decimal import ROUND_HALF_UP, Decimal from types import SimpleNamespace from typing import Any from sqlalchemy import and_, false, func, or_, select, update from sqlalchemy import null as sa_null from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.core.coordinators.bailian_cost_owner_coordinator import ( AUTO_OWNER_SOURCE_BAILIAN_KEY, bailian_api_key_id_from_resource_id, load_api_key_owner_map, ) from app.core.exceptions import format_exception_for_log from app.core.org_scope import resolve_department_subtree_names from app.core.safe_logging import get_safe_logger from app.domains.cmdb.cost_view import ( COST_MEASURE_INVOICE, engineering_row_clause, fact_amount_expr, non_engineering_row_clause, ) from app.domains.cmdb.models import ( TENANT_LEVEL_SCOPE_TYPES, CmdbBedrockProfileFact, CmdbCloudScope, CmdbCostAllocationRule, CmdbCostFact, ) from app.identity.person_models import IdentityPerson from app.integrations.driver_base import COST_BUCKET_USAGE, CostFact from app.integrations.models import Integration, PlatformSetting logger = get_safe_logger(__name__) # 6-decimal-place quantum matching the ``cmdb_cost_facts.amount_*`` # column scale (Numeric(14, 6)). Used at the float→Decimal boundary so # we don't round twice (once via Python ``round()``'s banker's # rounding, once via Numeric column truncation) and don't drift through # the float→str repr. _AMOUNT_QUANTUM = Decimal("0.000001") def _q6(value: Any) -> Decimal: """Quantize a float/Decimal/str amount to the 6-dp column scale via the str boundary (avoids double-rounding + float-repr drift). Mirrors the inline quantize the cost-facts upsert does.""" return Decimal(str(value or 0)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ) # ─── Ingestion ────────────────────────────────────────────────────── # NULL-component marker for ``compute_fact_key``. Chosen as the # null byte (NUL, U+0000) because no real provider string contains # it — guaranteeing zero collision risk with legitimate values # like ``"~"`` (an instance prefix used by some Aliyun bills) or # ``"-"``. Empty string is NOT used because Python's ``or`` would # also fold it onto the marker, but two facts with empty-string # components vs NULL components legitimately differ and must hash # distinctly. The driver contract states empty-string ≠ NULL — so # we treat them differently here too. _FACT_KEY_NULL_MARKER = "\x00" def compute_fact_key( *, integration_id: str, fact_date: date, scope_external_id: str | None, resource_id: str | None, usage_type: str | None, bucket_disambiguator: str | None = None, ) -> str: """sha256 hex of the natural key. Stable across re-ingestion. ``bucket_disambiguator`` (2026-07-28, cross-bucket collision fix): when a driver emits SAME-natural-key rows in DIFFERENT ``cost_bucket``\\s (aliyun same-day Refund/Adjustment sharing (date, scope, instance, item) with a usage row), merging them would misbucket the netted adjustment under the dominant member's bucket and corrupt the finance waterfall. The pre-aggregation layer keeps such rows separate and passes each non-``usage`` bucket's value here; it is appended as an extra hash component so the rows land as distinct DB rows. ``None`` — the ``usage`` bucket, which is the overwhelming majority of rows — keeps the historical 5-component hash. The 6-component string can never collide with a 5-component one because ``\\x01`` inside real components is escaped. The mapping bucket → disambiguator is :func:`_bucket_disambiguator` and is a pure function of the bucket alone; see its docstring for why batch-derived variants of it cannot converge. NULL components are collapsed to the U+0000 NUL marker so two "no-resource" rows on the same day for the same integration don't share a hash; the marker is collision-proof against any real-world provider string. Empty-string components are kept AS empty (not folded onto the marker) so the driver contract (empty ≠ NULL) holds end-to-end. ``scope_external_id`` is the cloud-side immutable identifier (Azure subscription id, GCP project id, Aliyun member uid, AWS linked-account id, Tencent ownerUin) — NOT the Helix-side ``cmdb_cloud_scopes.id`` UUID. The local DB id is mutable: it only exists after ``scopes_sync`` materialises the row, so a fact ingested before scopes_sync caught up would hash with NULL, and the same fact re-ingested after the catch-up would hash with a different scope_id and create a duplicate row. Anchoring to the cloud-side id is stable across that race. """ def _component(v: str | None) -> str: return _FACT_KEY_NULL_MARKER if v is None else v.replace("\x01", "\\x01") parts = [ fact_date.isoformat(), str(integration_id), _component(scope_external_id), _component(resource_id), _component(usage_type), ] if bucket_disambiguator is not None: parts.append("bucket:" + _component(bucket_disambiguator)) return hashlib.sha256("\x01".join(parts).encode("utf-8")).hexdigest() def _bucket_disambiguator(cost_bucket: str | None) -> str | None: """``cost_bucket`` → the ``compute_fact_key`` bucket component. A pure function of the bucket ALONE — deliberately independent of which buckets the current ingest batch happens to contain. ``usage`` owns the plain 5-component hash; every other bucket always carries its own name into the hash, whether or not a usage row for the same natural key is present in this batch. 2026-07-29: the first version of the cross-bucket fix elected the plain-hash keeper PER BATCH (usage if present, else the lexicographically-smallest bucket). That does not converge across re-ingests, because the bucket set for one natural key changes between syncs — both aliyun paths ``continue`` on ``native == 0``, so a day whose usage line nets to zero upstream drops out of the batch. The surviving adjustment row then moved back from its disambiguated hash to the plain hash, wrote a SECOND row, and orphaned its own previous one (nothing sweeps facts absent from a batch, so the duplicate is permanent: reproduced as a −30 refund stored twice, −60 total against an upstream truth of −30). Keying off the bucket alone makes every batch composition converge to the same rows. ``None`` / empty are non-canonical states no driver emits (``CostFact.cost_bucket`` defaults to ``usage``); they are mapped to distinct components so they can never collapse onto one another or onto ``usage``. """ if cost_bucket == COST_BUCKET_USAGE: return None if cost_bucket is None: return _FACT_KEY_NULL_MARKER return cost_bucket def _preaggregate_facts_by_fact_key( facts: list[CostFact], ) -> tuple[list[tuple[CostFact, str | None]], int]: """Collapse facts sharing a ``compute_fact_key`` natural key by SUMMING their monetary amounts — WITHIN one ``cost_bucket``. ``cmdb_cost_facts`` is keyed by ``fact_key`` = (date, integration, scope_external_id, resource_id, usage_type). A driver can legitimately emit MULTIPLE ``CostFact`` rows that map to the same fact_key because the natural key deliberately omits ``product_code`` / ``service`` (keeping the key provider-agnostic and migration-free). Concrete cases verified against prod (2026-06-17, BQ CUR + ``cmdb_cloud_billing_daily``): * **AWS** — CUR ``EdpDiscount`` / ``PrivateRateDiscount`` rows have a NULL ``line_item_resource_id`` but span 67 / 8 distinct ``product_code`` values. The driver synthesises their id from the *discount bucket* only (``aws-account::EDP Discount``), so every product_code collapses onto ONE fact_key. * **alicloud_intl** — BSS credit / true-up days emit several adjustment line items sharing a resource_id + billing-item. Pre-fix, the ON DUPLICATE KEY UPDATE upsert (no in-Python dedup) persisted only ONE colliding row per fact_key and silently DROPPED the rest. Because the dropped rows are disproportionately NEGATIVE (discounts / credits), the stored total OVER-reported spend: AWS ~+9% (+$370K / 30d) and alicloud_intl +$70K on a single credit day. ``cmdb_cloud_billing_daily`` was unaffected because it sums per-day in memory (no fact_key collapse). Summing within the fact_key grain is exactly correct: that grain IS the table's grain, so multiple same-grain rows must net rather than overwrite. Idempotent re-ingestion is preserved (same window → same sums → same overwrite). Provider-agnostic: a no-op for any driver whose facts already have unique fact_keys. The representative row (carrying the non-monetary metadata — service / region / tags / resource_name / raw) is the colliding member with the largest ``abs(amount_usd)`` so the dominant cost line names the merged cell; ties keep first-seen for determinism. Gross / usage_quantity follow the driver's "None means the provider didn't expose this" contract: summed across members that DID expose a value, left None only when NO member did. **Cross-bucket collisions never merge** (2026-07-28 fix): the merge key includes ``cost_bucket``, so an aliyun same-day Refund/Adjustment sharing (date, scope, instance, item) with a usage row stays a SEPARATE fact instead of netting into the usage row under the largest-|amount| member's bucket (which misbucketed adjustments and corrupted the finance waterfall / list whitelist). Because the DB unique key is the ``fact_key`` hash, the bucket must reach that hash: ``usage`` keeps the plain 5-component hash and every other bucket ALWAYS hashes with its bucket component appended (:func:`_bucket_disambiguator` — a pure function of the bucket, so the same fact converges to the same row no matter which other buckets a given batch carries; the 2026-07-29 revision of a per-batch keeper election that did NOT converge). Consequence for stored history: a non-``usage`` row written before 2026-07-29 sits under the plain hash and would never be rewritten again, double-counting against its new key on the next ingest. Alembic migration ``434`` re-keys exactly those rows (verified by recomputing the plain hash, so an unrecoverable natural key is skipped rather than guessed). Returns ``(collapsed, eliminated_count)`` where ``collapsed`` is a list of ``(fact, bucket_disambiguator | None)`` pairs to feed :func:`compute_fact_key`. """ # Accumulator keyed on the SAME natural key compute_fact_key hashes # (integration_id is constant within an ingest call, so it's omitted # from the in-memory key) PLUS cost_bucket so different-bucket rows # never merge. scope external_id mirrors the hash input. acc: dict[tuple, dict[str, Any]] = {} order: list[tuple] = [] for fact in facts: scope_ext = ( fact.scope.external_id if fact.scope is not None else None ) key = ( fact.date, scope_ext, fact.resource_id, fact.usage_type, fact.cost_bucket, ) rec = acc.get(key) if rec is None: rec = { "rep": fact, "rep_abs": abs(fact.amount_usd or 0.0), "amount_usd": 0.0, "amount_native": 0.0, "gross_usd": 0.0, "gross_usd_seen": False, "gross_native": 0.0, "gross_native_seen": False, "list_usd": 0.0, "list_usd_seen": False, "list_native": 0.0, "list_native_seen": False, "public_usd": 0.0, "public_usd_seen": False, "usage_qty": 0.0, "usage_qty_seen": False, "currencies": set(), "count": 0, } acc[key] = rec order.append(key) # ``amount_native`` is summed in the row's native currency, but # ``fact_key`` (and this grouping key) omit currency. Summing two # different-currency facts into one ``amount_native`` would stamp # the merged total with a single currency label and silently # corrupt native-currency analytics. Every current driver guards # this upstream (AWS BQ/CE raise on mixed-currency collisions; # aliyun/tencent/ucloud are single-currency per integration; # gcp/azure/alicloud_intl convert before emit) — but defend the # invariant here too so a future driver dropped into this path # fails loudly instead of producing wrong native totals. # ``amount_usd`` is always safe to sum (non-USD rows carry # amount_usd=0.0 per the CostFact contract). if fact.currency: rec["currencies"].add(fact.currency) rec["amount_usd"] += float(fact.amount_usd or 0.0) rec["amount_native"] += float(fact.amount_native or 0.0) if fact.amount_usd_gross is not None: rec["gross_usd"] += float(fact.amount_usd_gross) rec["gross_usd_seen"] = True if fact.amount_native_gross is not None: rec["gross_native"] += float(fact.amount_native_gross) rec["gross_native_seen"] = True # 双口径:list / public 与 gross 同一套 None≠0 语义。 if fact.amount_usd_list is not None: rec["list_usd"] += float(fact.amount_usd_list) rec["list_usd_seen"] = True if fact.amount_native_list is not None: rec["list_native"] += float(fact.amount_native_list) rec["list_native_seen"] = True if fact.amount_usd_public is not None: rec["public_usd"] += float(fact.amount_usd_public) rec["public_usd_seen"] = True if fact.usage_quantity is not None: rec["usage_qty"] += float(fact.usage_quantity) rec["usage_qty_seen"] = True rec["count"] += 1 # Largest-magnitude member wins the metadata slot (strict > so # ties keep first-seen → deterministic regardless of input order). cur_abs = abs(fact.amount_usd or 0.0) if cur_abs > rec["rep_abs"]: rec["rep"] = fact rec["rep_abs"] = cur_abs _mixed = {k: acc[k]["currencies"] for k in order if len(acc[k]["currencies"]) > 1} if _mixed: _ex_k, _ex_cs = next(iter(_mixed.items())) raise RuntimeError( "cost_facts pre-aggregation: facts sharing one fact_key span " "multiple currencies (compute_fact_key omits currency, so they " "would merge into one amount_native with a single currency " f"label). Example date={_ex_k[0]} scope={_ex_k[1]} " f"resource={_ex_k[2]} currencies={sorted(_ex_cs)}. The driver " "must convert to a single currency per fact_key (or currency " "must be added to the fact_key natural key + UNIQUE migration)." ) out: list[tuple[CostFact, str | None]] = [] eliminated = 0 for key in order: rec = acc[key] eliminated += rec["count"] - 1 rep: CostFact = rec["rep"] # Bucket → hash component is a pure function of the bucket (NOT # of this batch's bucket set) so re-ingest converges for any # batch composition. See _bucket_disambiguator. bucket_disambiguator = _bucket_disambiguator(key[4]) out.append(( _dc_replace( rep, amount_usd=rec["amount_usd"], amount_native=rec["amount_native"], amount_usd_gross=( rec["gross_usd"] if rec["gross_usd_seen"] else None ), amount_native_gross=( rec["gross_native"] if rec["gross_native_seen"] else None ), amount_usd_list=( rec["list_usd"] if rec["list_usd_seen"] else None ), amount_native_list=( rec["list_native"] if rec["list_native_seen"] else None ), amount_usd_public=( rec["public_usd"] if rec["public_usd_seen"] else None ), usage_quantity=( rec["usage_qty"] if rec["usage_qty_seen"] else None ), ), bucket_disambiguator, )) return out, eliminated # Maximum serialised size of the per-fact ``raw`` audit payload. # SQLAlchemy JSON column on TiDB stores up to ~64 KiB before TEXT # truncation; we apply a string-length bound after serialisation to # keep ingestion well clear of the row-size limit even on chunked # multi-row inserts. Provider payloads larger than this are dropped # rather than truncated to invalid JSON — only the audit context is # lost; the typed CostFact columns carry the analytics-relevant data. _RAW_PAYLOAD_MAX_BYTES = 16_000 # Per-scope timeout for ``cmdb.cost_facts_sync``. A single hung driver # call MUST NOT block the rest of the cron — without this guard the # whole run wedges in ``running`` state until manually intervened # (2026-05-19 prod incident). 300s is generous enough for any single # AWS account's worst-case CUR-row volume; tests monkey-patch to # ~0.2s to keep the suite fast. # # Note on thread cleanup: when this asyncio timeout fires, the # awaitable is cancelled but the driver's underlying sync work # (running inside ``asyncio.to_thread``) keeps executing until BQ # itself returns. The AwsCloudDriver's BQ jobs carry an explicit # ``job_timeout_ms = 5 * 60 * 1000`` (5 min) in # ``_bq_query_job_config``, so the orphan-thread lifetime is bounded # to the same 5 minutes — no unbounded resource leak. _COST_FACTS_PER_SCOPE_TIMEOUT_SEC: float = 300.0 # Bedrock profile-facts sync fans out per (axis × spending-key) — each CE # call is ~1s in prod, but the whole per-scope leg had NO timeout while # holding the shared ``cloud_sync:cost_facts:`` lock (ttl=900), so a # wedged CE call could keep the lock until expiry and let the next cron # tick run a concurrent DELETE+upsert on the same window. Bound it well # above the realistic call count (single-digit keys × a couple of axes) # yet far below the lock TTL. _BEDROCK_PROFILE_PER_SCOPE_TIMEOUT_SEC: float = 120.0 # Which configured ``aws_profile_tag`` axes are per-API-Key principal # tags that should be CROSSED with the ``UsedBy`` workload tag (migration # 452) instead of pulled owner-fan-out style. Mirrors # ``apikey.spend.PER_KEY_PROFILE_AXES`` (kept local to avoid a cmdb→apikey # domain import); the two lists MUST stay in sync — a value read as # per-key there but pulled workload-style here would leave the per-key # detail page with no ``sub_profile`` rows. _PER_KEY_PRINCIPAL_AXES: frozenset[str] = frozenset({ "iamprincipal/helix-request-number", "helix-request-number", }) # The workload tag the per-key axis is sub-divided by when the configured # list doesn't name one explicitly. _DEFAULT_SUB_PROFILE_TAG = "UsedBy" def _safe_json_payload(raw: dict | None) -> dict: """Return a JSON-safe, size-bounded copy of a raw provider payload. The driver-supplied ``raw`` dict can carry arbitrary types (``Decimal``, ``datetime``, boto3 ``botocore.response`` streams, Tencent SDK objects with ``__repr__`` -only serialisation). We coerce everything to a JSON-roundtrippable shape via ``json.dumps(..., default=str)``, bound the size, and reload so the SQLAlchemy JSON column stores a real dict (not an over-encoded string). Returns ``{}`` when serialisation fails or the payload is None. """ if not raw: return {} try: encoded = json.dumps(raw, ensure_ascii=False, default=str) except (TypeError, ValueError): return {} if len(encoded) > _RAW_PAYLOAD_MAX_BYTES: # Bound rather than truncate — truncated JSON is invalid # JSON, and an invalid JSON value would raise on insert. A # marker dict keeps the audit row honest. Marker keys use # a `__helix_` prefix so a real provider payload happening # to carry a literal `_truncated` key can't be confused for # our marker downstream. return { "__helix_truncated__": True, "__helix_raw_size_bytes__": len(encoded), "__helix_raw_max_bytes__": _RAW_PAYLOAD_MAX_BYTES, } try: return json.loads(encoded) except (TypeError, ValueError): return {} async def _upsert_with_adaptive_split( session: AsyncSession, rows: list[dict[str, Any]], ) -> int: """Run an UPSERT for ``rows`` and recover from TiDB memory-quota errors by recursively halving the batch. Background — TiDB's ``tidb_mem_quota_query`` (default 1 GiB) caps the executor working-set for a single statement, NOT just the wire payload. For ``cmdb_cost_facts`` with JSON columns (``raw`` up to 16 KiB, ``tags`` dict) + ON DUPLICATE KEY UPDATE row-lookups, AWS BigQuery CUR rows occasionally tip past the cap on the largest accounts even at 100 rows/batch. Without this helper the whole batch is lost (no fallback path); with it we bisect until each sub-batch fits, so a single oversized row (or pathological JSON) only loses the singleton row, not the rest. Returns the number of rows successfully upserted. A row that fails even at batch_size=1 is logged + skipped; the function never raises ``OperationalError(8175)`` — only re-raises other errors that the caller (cron handler) is set up to handle. Pinned by ``test_upsert_with_adaptive_split_halves_on_mem_quota``. """ from sqlalchemy.exc import OperationalError if not rows: return 0 stmt = mysql_insert(CmdbCostFact.__table__).values(rows) stmt = stmt.on_duplicate_key_update( scope_id=stmt.inserted.scope_id, resource_group=stmt.inserted.resource_group, resource_name=stmt.inserted.resource_name, resource_type=stmt.inserted.resource_type, region=stmt.inserted.region, service=stmt.inserted.service, tags=stmt.inserted.tags, amount_usd=stmt.inserted.amount_usd, amount_native=stmt.inserted.amount_native, amount_usd_gross=stmt.inserted.amount_usd_gross, amount_native_gross=stmt.inserted.amount_native_gross, amount_usd_list=stmt.inserted.amount_usd_list, amount_native_list=stmt.inserted.amount_native_list, amount_usd_public=stmt.inserted.amount_usd_public, cost_bucket=stmt.inserted.cost_bucket, currency=stmt.inserted.currency, usage_quantity=stmt.inserted.usage_quantity, usage_unit=stmt.inserted.usage_unit, cost_owner_person_id=stmt.inserted.cost_owner_person_id, cost_owner_department=stmt.inserted.cost_owner_department, cost_center=stmt.inserted.cost_center, allocation_rule_id=stmt.inserted.allocation_rule_id, auto_owner_source=stmt.inserted.auto_owner_source, raw=stmt.inserted.raw, updated_at=func.now(), ) try: await session.execute(stmt) return len(rows) except OperationalError as exc: # TiDB error 8175: ``tidb_mem_quota_query`` exceeded. # asyncmy surfaces this as ``OperationalError`` with (8175, ...) # as args[0] — pattern-match instead of relying on a vendor- # specific exception class so we don't couple to asyncmy. msg = str(exc) is_mem_quota = "8175" in msg or "tidb_mem_quota_query" in msg if not is_mem_quota: # Other OperationalErrors (deadlock, conn loss, etc.) — # let the caller's outer try/except handle them. raise # Rollback this statement so the session is usable for the # halved retries below. Without rollback the next execute # would PendingRollbackError. await session.rollback() if len(rows) == 1: # Even a singleton row exceeded — almost always a giant # ``raw`` JSON or an exotic tag dict. Log + drop the row # rather than infinite-recursing. The fact_key is logged # so an operator can hunt it down post-hoc. logger.warning( "cost_facts: dropping row that exceeds " "tidb_mem_quota_query even at batch_size=1: " "fact_key=%s integration=%s resource=%s service=%s", rows[0].get("fact_key"), rows[0].get("integration_id"), rows[0].get("resource_id"), rows[0].get("service"), ) return 0 # Recursive halve. Both halves get the same retry logic. # Commit between halves so a rollback() inside the second half's # OOM handler cannot undo the first half's already-succeeded rows. # Without this, rollback() operates at the full-session level and # silently drops the first half when the second half recurses. mid = len(rows) // 2 first = await _upsert_with_adaptive_split(session, rows[:mid]) await session.commit() second = await _upsert_with_adaptive_split(session, rows[mid:]) return first + second async def ingest_cost_facts( session: AsyncSession, *, integration_id: str, facts: list[CostFact], scope_lookup: dict[str, str] | None = None, ) -> dict[str, int]: """Upsert a batch of CostFact rows into ``cmdb_cost_facts``. Allocation columns are stamped from the rule ladder at ingest time; if no rule matches, the row is inserted with NULL owner/cc and surfaces in the "未分摊" KPI. ``scope_lookup`` maps ``scope.external_id`` → ``cmdb_cloud_scopes.id`` so the driver-emitted ``CloudScopeRef`` (no DB id) can be joined back to the local scope row. Caller is responsible for building this; the cron handler does it once per integration. Returns a dict of ``{"upserted": N, "skipped": M}`` for the scheduled_job_runs row's summary field. """ if not facts: return {"upserted": 0, "skipped": 0} # Collapse driver-emitted facts that share a fact_key (summing # amounts) BEFORE allocation + the chunked upsert. Without this, the # ON DUPLICATE KEY UPDATE upsert keeps only one colliding row per # fact_key and silently drops the rest — disproportionately the # NEGATIVE discount / credit lines, over-reporting spend. Done once, # up front, so the dedup also spans chunk boundaries (two colliding # facts in different 100-row batches would otherwise still overwrite # each other). See _preaggregate_facts_by_fact_key for the prod # evidence (AWS +9% / alicloud_intl credit-day +$70K, 2026-06-17). # Cross-bucket collisions do NOT merge — the second tuple member is # the ``bucket_disambiguator`` threaded into ``compute_fact_key`` # below so different-bucket rows persist as distinct facts. fact_entries, collapsed_count = _preaggregate_facts_by_fact_key(facts) if collapsed_count: logger.info( "cost_facts ingest %s: pre-aggregated %d colliding fact(s) " "into shared fact_keys (summed amounts; prevents the upsert " "from dropping co-located discount/credit rows)", integration_id, collapsed_count, ) rules = await _load_active_rules(session) # Resolved once per ingest, not per row — see the "raw" key below. _raw_cutoff = datetime.now(UTC).date() - timedelta( days=await resolve_raw_retention_days(session), ) integration = await session.get(Integration, uuid.UUID(integration_id)) if integration is None: # Refuse to ingest against a missing integration — provider # would silently default to "" and every provider-conditional # rule would skip, leaving facts unallocated for reasons that # don't surface in any UI. Caller (cron handler) catches this # and records the failure against the integration. raise RuntimeError( f"ingest_cost_facts: integration {integration_id} not found", ) provider = integration.provider or "" # Pre-launch hardening 2026-05-16 follow-up (LS1 P1): AWS-only toggle # guard. ``billing_source`` (``cost_explorer`` vs ``bigquery``) is an # admin-tunable on the AWS integration's config_public. Pre-fix, an # admin flipping the source between cron ticks left two parallel # ingestion paths writing rows with different ``usage_type`` shapes # (CE emits ``None``; BQ emits the real ``line_item_usage_type``) — # ``compute_fact_key`` includes ``usage_type`` so the same logical # day×resource produced TWO ``cmdb_cost_facts`` rows and ``SUM # (amount_usd)`` double-counted that window. We compare the current # ``billing_source`` to the marker we last stamped after a successful # ingest; on mismatch we raise loudly and surface the recommended # purge SQL so the admin can clear the conflicting window before # re-running. Skips entirely for non-AWS integrations. # # Round-1 review follow-up (Review-C P2-1): an empty-string marker # value (left over from an out-of-band write or a botched legacy # row) is normalised to ``None`` so we don't fire a spurious # "toggle" error against a known-good ``cost_explorer`` config. current_source_for_marker: str | None = None if provider == "aws": current_source = str( (integration.config_public or {}).get("billing_source") or "cost_explorer", ) current_source_for_marker = current_source marker_key = ( f"cost_facts.last_billing_source.{integration_id}" ) marker_row = await session.get(PlatformSetting, marker_key) prior_source: str | None = None if marker_row is not None and isinstance(marker_row.value, str): stripped = marker_row.value.strip() if stripped: prior_source = stripped if prior_source is not None and prior_source != current_source: raise RuntimeError( f"ingest_cost_facts: AWS integration {integration.name!r} " f"billing_source changed from {prior_source!r} → " f"{current_source!r}; refusing to ingest until the operator " "purges the previously-ingested cost facts for this integration " "and resets the billing_source marker via the CMDB admin panel " "or by contacting the DBA. Refer to ops runbook for details." ) upserted = 0 skipped = 0 missing_scopes: set[str] = set() # Track whether the AWS billing_source marker has been upserted # in this ingest; the LS4 P2 fix piggy-backs the marker write on # the FIRST batch commit so they share atomicity. marker_written: bool = False # Pre-load the scope→owner map for this tenant so the fallback # chain (rule miss → scope owner) doesn't issue per-fact queries. # Empty dict when no scope has an owner — falls cleanly through # to ``auto_owner_source='none'``. scope_owner_lookup = await _build_scope_owner_lookup( session, integration_id=integration_id, ) # 百炼 Key → 人。小表全量读一次(生产 31 行);Python 侧查表而非 # SQL JOIN,因为两张表的 collation 不同(见 coordinator docstring)。 bailian_key_owners = await load_api_key_owner_map(session) person_meta_cache = await _build_person_meta_cache( session, rules, extra_person_ids=( set(scope_owner_lookup.values()) | set(bailian_key_owners.values()) ) or None, ) # Chunk to keep the executemany payload bounded. # # Default 100 rows/batch (was 500): every fact carries a JSON # ``raw`` (up to 16 KiB) + ``tags`` dict + ~20 typed columns, so # 500 rows × ~17 KiB worst-case = ~8 MiB per INSERT, and TiDB's # default ``tidb_mem_quota_query`` (1 GiB) measures the executor's # working-set during INSERT planning + ON DUPLICATE KEY UPDATE # row-lookup, not just the bytes on the wire — for fat AWS BQ CUR # payloads with hundreds of ``resource_tags_user_*`` columns it # actually exceeded the cap in prod (2026-05-16, ``Your query has # been cancelled due to exceeding the allowed memory limit`` on # every batch of the AWS main account historical backfill; # ~5 SQL errors silently dropped 6 month × N-account cost_facts). # 100 rows/batch keeps the working-set well below 1 GiB even on # the fattest CUR shapes. chunk_size = 100 for batch_start in range(0, len(fact_entries), chunk_size): batch = fact_entries[batch_start: batch_start + chunk_size] rows: list[dict[str, Any]] = [] for fact, bucket_disambiguator in batch: scope_db_id: str | None = None if fact.scope and fact.scope.external_id: scope_db_id = (scope_lookup or {}).get( fact.scope.external_id, ) # Track scopes the driver mentioned that we didn't # have a Helix-side row for. Surface in summary so # the operator knows scopes_sync is lagging the # cost-facts feed (rather than silently writing # facts with NULL scope_id, which makes the # by-scope drilldown lie). if scope_db_id is None: missing_scopes.add(fact.scope.external_id) # Hash on the cloud-side external_id, NOT the Helix UUID # (scope_db_id). The DB id is materialised by scopes_sync; # a fact ingested before scopes_sync would hash with # scope=NULL and re-ingested after the catch-up would # hash with scope=, creating a duplicate row instead # of an idempotent upsert. fact_key = compute_fact_key( integration_id=integration_id, fact_date=fact.date, scope_external_id=( fact.scope.external_id if fact.scope else None ), resource_id=fact.resource_id, usage_type=fact.usage_type, bucket_disambiguator=bucket_disambiguator, ) # Savings Plan discount / fee rows are NOT user-attributable # spend — they are an org-level commitment netting / cost # against the entire scope. Allocating them to a scope # owner makes that owner's per-person chargeback view # carry a giant negative (linked-account SP negation) or # positive (payer SP recurring/upfront fee) line that is # not their own activity. Leave SP rows unallocated so # they sum into a clearly-labelled "Unallocated" bucket on # per-person / per-department breakdowns, while still # contributing to per-scope totals (which is the level # where the SP discount belongs by AWS's billing model). # Caught by round-2 review. is_sp_row = fact.service == "Savings Plans" if is_sp_row: allocation = {} cost_owner_person_id = None cost_owner_department = None cost_center = None allocation_rule_id = None auto_owner_source = "savings_plan" else: allocation = _match_rule( rules, provider=provider, integration_id=integration_id, resource_group=fact.resource_group, service=fact.service, tags=fact.tags or {}, ) cost_owner_person_id = allocation.get("person_id") cost_owner_department = allocation.get("department") cost_center = allocation.get("cost_center") allocation_rule_id = allocation.get("rule_id") auto_owner_source = "rule" if allocation_rule_id else "none" # 百炼 API Key 归属 —— 排在规则之后、订阅 Owner 兜底之前。 # BSS 把 API Key id 编码在 ``resource_id`` 首段,阶梯此前 # 认不出这个形状,于是整个百炼账号的钱全落到账号 Owner 一个 # 人头上(生产 2026-08:$8,100.94 全记在一人名下,其中 # $4,272.39 其实能定位到 11 位同事)。解析不出人的行继续往下 # 走兜底 —— 资源目录成员账号里的 Key 无法枚举,那部分残留是 # 真实的,详见 coordinator docstring。 if not cost_owner_person_id and not is_sp_row: bailian_key_id = bailian_api_key_id_from_resource_id( fact.service, fact.resource_id, ) if bailian_key_id: bailian_owner = bailian_key_owners.get(bailian_key_id) if bailian_owner: cost_owner_person_id = bailian_owner auto_owner_source = AUTO_OWNER_SOURCE_BAILIAN_KEY # Fallback: when no rule matched, inherit the fact's # ``cmdb_cloud_scopes.owner_person_id`` (the default # owner of the subscription / project / sub-account # the fact lives in). This is the chain the user # described — every dollar lands on a billable scope, # every active scope has an owner, every owner has a # primary org membership → therefore default attribution # works without a single admin-authored rule. Rules # remain the override layer (e.g. an Azure RG split off # a subscription to a different applicant via the # apikey auto-grant flow). SP discount rows skip the # fallback too (see ``is_sp_row`` above). if ( not cost_owner_person_id and scope_db_id and not is_sp_row ): fallback_person = scope_owner_lookup.get(scope_db_id) if fallback_person: cost_owner_person_id = fallback_person auto_owner_source = "scope_owner" # Resolve dept + cost-center from the person's primary org # membership when neither the rule nor the fallback # carried explicit values. The cache is pre-warmed for # every rule-referenced person AND every scope-owner so # large ingestion batches don't N+1. if cost_owner_person_id: meta = person_meta_cache.get(cost_owner_person_id) or ( None, None, ) if cost_owner_department is None: cost_owner_department = meta[0] if cost_center is None: cost_center = meta[1] # Round-49 pre-launch hardening: reject NaN / ±Inf at the # boundary BEFORE the Decimal conversion. ``Decimal('NaN')`` # and ``Decimal('Infinity')`` are valid Python objects but # MySQL/TiDB ``DECIMAL(14,6)`` rejects them on INSERT; the # adaptive-split path in ``_upsert_with_adaptive_split`` # would then halve down to batch_size=1, log a generic # "tidb_mem_quota_query" warning (the only branch that # silently drops a row) — except the failure mode here is # an arithmetic conversion error (1366 / 1265), not a memory # error, so the singleton-drop branch doesn't even catch it # and the entire ingest aborts mid-batch. Either way the # operator gets zero per-fact context. Raise loudly with # the fact_key so a misbehaving driver is debuggable instead # of silently corrupting the cost surface. for _field_name, _value in ( ("amount_usd", fact.amount_usd), ("amount_native", fact.amount_native), ("amount_usd_gross", fact.amount_usd_gross), ("amount_native_gross", fact.amount_native_gross), ): # gross fields are optional — skip the finiteness gate # when the provider didn't expose a list price. if _value is not None and not math.isfinite(_value): raise RuntimeError( "ingest_cost_facts: rejecting non-finite " f"{_field_name}={_value!r} for fact_key=" f"{fact_key} integration={integration_id} " f"date={fact.date} resource={fact.resource_id!r}" ) if ( fact.usage_quantity is not None and not math.isfinite(fact.usage_quantity) ): raise RuntimeError( "ingest_cost_facts: rejecting non-finite " f"usage_quantity={fact.usage_quantity!r} for " f"fact_key={fact_key} integration={integration_id}" ) # ``tags`` and ``raw`` are SQLAlchemy JSON columns — # they want a Python dict/list, not a pre-serialised # string (which would be re-encoded into a JSON string, # double-encoding the data into the column). ``raw`` is # additionally walked through ``_safe_json_payload`` to # ensure every nested value is JSON-serialisable # (Decimals, datetimes, the boto3 botocore.response # streaming types) and bounded so an exotic provider # payload can't blow past the row size limit. rows.append({ "fact_key": fact_key, "date": fact.date, "integration_id": integration_id, "scope_id": scope_db_id, "resource_group": fact.resource_group, "resource_id": fact.resource_id, "resource_name": fact.resource_name, "resource_type": fact.resource_type, "region": fact.region, "service": fact.service or "Other", "usage_type": fact.usage_type, "tags": dict(fact.tags or {}), # Round-27 pre-launch (2026-05-25): replace # ``Decimal(str(round(float, 6)))`` with quantize. The # old form went float → ``round()`` (banker's rounding) # → ``str()`` (which can emit 1.2345670000000001 shapes) # → ``Decimal`` — a measurable sub-microcent drift per # row at penny-precise reconcile. Quantize keeps the # Decimal symbolic until the column-bound 6dp scale. "amount_usd": Decimal(str(fact.amount_usd)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ), "amount_native": Decimal(str(fact.amount_native)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ), # 折扣前(原价)。None 透传(provider 无原价口径), # 大盘按覆盖率聚合,不伪造折扣。 "amount_usd_gross": ( Decimal(str(fact.amount_usd_gross)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ) if fact.amount_usd_gross is not None else None ), "amount_native_gross": ( Decimal(str(fact.amount_native_gross)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ) if fact.amount_native_gross is not None else None ), # 双口径:业务折前成本(研发口径)/ 按需牌价。None 透传 # (=该行不在白名单或该云口径未接入)——读侧裸 SUM, # 绝不把 None 写成 0。 "amount_usd_list": ( Decimal(str(fact.amount_usd_list)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ) if fact.amount_usd_list is not None else None ), "amount_native_list": ( Decimal(str(fact.amount_native_list)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ) if fact.amount_native_list is not None else None ), "amount_usd_public": ( Decimal(str(fact.amount_usd_public)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ) if fact.amount_usd_public is not None else None ), "cost_bucket": fact.cost_bucket or "usage", "currency": fact.currency, "usage_quantity": ( Decimal(str(fact.usage_quantity)).quantize( _AMOUNT_QUANTUM, rounding=ROUND_HALF_UP, ) if fact.usage_quantity is not None else None ), "usage_unit": fact.usage_unit, "cost_owner_person_id": cost_owner_person_id, "cost_owner_department": cost_owner_department, "cost_center": cost_center, "allocation_rule_id": allocation_rule_id, "auto_owner_source": auto_owner_source, # Retention guard (2026-09-01): never persist ``raw`` # for a day the purge cron has already swept past. # Without this a wide manual backfill would re-populate # ``raw`` for months that sit BELOW the purge watermark, # where nothing would ever collect it again. ``raw`` has # no readers, so a backfill losing it costs nothing. "raw": ( _safe_json_payload(fact.raw) if fact.date >= _raw_cutoff else None ), }) if not rows: continue upserted += await _upsert_with_adaptive_split(session, rows) # Round-3 review follow-up (LS4 P2): the AWS billing_source # marker must commit in the SAME transaction as the first # batch of facts. Pre-fix, the marker upsert + commit happened # at the end of ingest, so a failure between the LAST batch # commit and the marker commit (network blip, TiDB tick-down) # left facts persisted with the NEW source but the marker # still pointing at the OLD source. The toggle guard on the # next ingest would then false-positive and block all future # ingests until the operator ran the suggested-purge SQL. # Now: write the marker just before the FIRST batch's commit # so they share atomicity. Subsequent batches see # ``marker_written=True`` and skip — single write per ingest. if provider == "aws" and not marker_written: from sqlalchemy.dialects.mysql import insert as mysql_insert marker_key = ( f"cost_facts.last_billing_source.{integration_id}" ) await session.execute( mysql_insert(PlatformSetting) .values( key=marker_key, value=current_source_for_marker, description=( "Last billing_source observed during a " "successful AWS cost_facts ingest. Cron-" "managed; do NOT edit via the admin UI — see " "docs/ops/cloud-lifecycle-ops.md §9a-aws-bq " "for the toggle procedure." ), ) .on_duplicate_key_update(value=current_source_for_marker) ) marker_written = True # Commit per batch so a downstream batch failure can't roll # back already-written progress. Pre-fix the commit was at # the end of the whole ingest call, so one TiDB # ``tidb_mem_quota_query`` rejection mid-loop discarded every # earlier batch in the same call — turning a recoverable # partial write into a wholesale data loss for the # (integration, window) tuple. Idempotent ``fact_key`` makes # re-running the whole ingest safe regardless; per-batch # commit just preserves forward progress. await session.commit() if missing_scopes: logger.warning( "cost_facts ingest %s: %d driver-side scope id(s) had no " "matching cmdb_cloud_scopes row — facts written with " "scope_id=NULL: %s", integration_id, len(missing_scopes), sorted(missing_scopes)[:10], ) # Round-3 review follow-up (LS4 P2) — end-of-loop fallback: every # batch had its rows filtered out (rule mismatch / NULL scope_id # only / skipped) so we never committed anything inside the loop. # Still stamp the marker — admin already chose this source by # configuring it, and "ingest ran, produced 0 persisted rows" is # a state the toggle guard must observe. Without this, an entire # window of empty ingests after a toggle would leave the marker # stuck on the old value, false-positive on the next real ingest. # # Round-41 follow-up (2026-05-19): gate on ``upserted > 0`` rather # than ``facts`` (truthy input list). Pre-fix, a run that 8175- # dropped every batch in ``_upsert_with_adaptive_split`` even at # batch_size=1 still stamped the marker as if it succeeded — the # toggle guard then went BLIND on the next ingest (would not flag # a real source switch because the prior_source equals the new # marker that came from a zero-write run). Tying the marker write # to actual persistence keeps the guard's invariant honest. if ( provider == "aws" and not marker_written and current_source_for_marker is not None and upserted > 0 ): from sqlalchemy.dialects.mysql import insert as mysql_insert marker_key = ( f"cost_facts.last_billing_source.{integration_id}" ) await session.execute( mysql_insert(PlatformSetting) .values( key=marker_key, value=current_source_for_marker, description=( "Last billing_source observed during a successful " "AWS cost_facts ingest. Cron-managed; do NOT edit " "via the admin UI — see " "docs/ops/cloud-lifecycle-ops.md §9a-aws-bq for " "the toggle procedure." ), ) .on_duplicate_key_update(value=current_source_for_marker) ) await session.commit() return { "upserted": upserted, "skipped": skipped, "collapsed": collapsed_count, "missing_scopes": sorted(missing_scopes), } async def build_scope_lookup( session: AsyncSession, *, integration_id: str, ) -> dict[str, str]: """Map ``scope.external_id`` → ``cmdb_cloud_scopes.id`` for one tenant. Round-22: when two rows legitimately share the same ``external_id`` (e.g. a soft-delete + recreate sequence where some legacy callers used hard-delete instead of decommission, or a re-import after rollback), the iteration ordering of the result set is undefined on TiDB. Without a deterministic tiebreaker, facts can flip-flop between the two scope_ids run-to-run depending on which row the cursor surfaces first. Anchor on the active row when present, then on the older ``created_at`` so we get reproducible attribution. The trailing ``id`` is purely for total determinism when timestamps collide. """ res = await session.execute( select( CmdbCloudScope.id, CmdbCloudScope.external_id, CmdbCloudScope.deleted_at, CmdbCloudScope.created_at, ).where( CmdbCloudScope.integration_id == integration_id, CmdbCloudScope.deleted_at.is_(None), ) .order_by( CmdbCloudScope.created_at.asc(), CmdbCloudScope.id.asc(), ), ) out: dict[str, str] = {} for row in res: # First write wins — earliest ``created_at`` is canonical. out.setdefault(row.external_id, str(row.id)) return out async def _build_scope_owner_lookup( session: AsyncSession, *, integration_id: str | None = None, ) -> dict[str, str]: """Map ``cmdb_cloud_scopes.id`` (string) → ``owner_person_id`` for the scope-owner attribution fallback. Powers the "no rule matched, but the scope has a known owner" path: a fact whose ``scope_id`` resolves to an active scope row with an owner inherits that owner. Concrete chain example: Azure subscription "tidb-o11y" → owner=lu.xing@pingcap.com → person.primary_org_membership leaf = "O11Y" → fact.cost_owner_department = "O11Y" Filtered to ``deleted_at IS NULL`` because soft-deleted scopes can have stale owners (the person may have left or moved teams) and re-attributing fresh facts to a departed owner produces a misleading dashboard. Facts pointing at a soft-deleted scope fall through to ``auto_owner_source='none'``. ``integration_id`` filter exists for the ingest path which processes one tenant at a time; the recompute path passes None and walks every active scope across every cloud. Returns empty dict when no scope has an owner — caller should treat that as "no fallback available", not "skip ingestion". """ q = select( CmdbCloudScope.id, CmdbCloudScope.owner_person_id, ).where( CmdbCloudScope.owner_person_id.isnot(None), CmdbCloudScope.deleted_at.is_(None), ) if integration_id: q = q.where(CmdbCloudScope.integration_id == integration_id) res = await session.execute(q) return { str(r.id): str(r.owner_person_id) for r in res if r.owner_person_id } # ─── Allocation ───────────────────────────────────────────────────── async def _build_person_meta_cache( session: AsyncSession, rules: list[CmdbCostAllocationRule], *, extra_person_ids: set[str] | None = None, ) -> dict[str, tuple[str | None, str | None]]: """Pre-compute ``(department_leaf, cost_center)`` for every person any rule references AND every scope-owner that may be reached via the fallback chain. Department leaf is derived from the primary ``identity_org_memberships`` row's ``org_path`` (``leaf-most node after stripping the tenant root``) — same convention :meth:`IdentityPerson.department_chain` uses. Cost center is the direct column on the person row (populated by the SF EmpJob.costCenter mapping). We query the join directly rather than touching the ``org_memberships`` relationship so this works in async sessions without selectinload gymnastics. ``extra_person_ids`` carries the union of every ``owner_person_id`` referenced by ``cmdb_cloud_scopes`` so the ingest path can resolve dept/CC for scope-owner-fallback attributions in the same single query (no N+1 per fact). Without this parameter the cache only holds rule-targeted persons, which means the fallback path would have to issue one extra query per fact to resolve dept/CC. Returns ``{person_id: (department_leaf, cost_center)}``. """ person_ids = { str(r.assign_person_id) for r in rules if r.assign_person_id } if extra_person_ids: person_ids.update(extra_person_ids) if not person_ids: return {} # Avoid an import cycle — IdentityOrgMembership lives in the same # module as IdentityPerson. from app.identity.person_models import IdentityOrgMembership person_rows = await session.execute( select(IdentityPerson.id, IdentityPerson.cost_center) .where(IdentityPerson.id.in_(person_ids)), ) persons = {str(r.id): r.cost_center for r in person_rows} member_rows = await session.execute( select( IdentityOrgMembership.person_id, IdentityOrgMembership.org_path, IdentityOrgMembership.org_node_name, ) .where( IdentityOrgMembership.person_id.in_(person_ids), IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ), ) dept_by_person: dict[str, str | None] = {} for r in member_rows: raw = r.org_path or r.org_node_name or "" if not raw: continue parts = [p.strip() for p in raw.split("/") if p.strip()] if parts and parts[0].lower() in {"pingcap", "pingcap-ati"}: parts = parts[1:] if parts: dept_by_person[str(r.person_id)] = parts[-1] elif r.org_node_name: dept_by_person[str(r.person_id)] = r.org_node_name return { pid: (dept_by_person.get(pid), cc) for pid, cc in persons.items() } async def _load_active_rules( session: AsyncSession, ) -> list[CmdbCostAllocationRule]: # Stable secondary tiebreaker: when two rules share a priority, # ``priority.asc()`` alone leaves the order implementation-defined, # so the per-fact "first match wins" can flip across ingest runs # vs. ``recompute_allocation`` runs on the same data — the same # fact's allocation_rule_id silently swaps between two rules with # the same priority. Add ``created_at.asc()`` so the *older* rule # wins on ties (matches operator intuition: rules created first # take precedence; new rules with the same priority slot in # afterwards). ``id.asc()`` as a final UUID-monotonic-ish tiebreaker # so even rules created in the same millisecond have a stable # winner. res = await session.execute( select(CmdbCostAllocationRule) .where(CmdbCostAllocationRule.enabled.is_(True)) .order_by( CmdbCostAllocationRule.priority.asc(), CmdbCostAllocationRule.created_at.asc(), CmdbCostAllocationRule.id.asc(), ), ) return list(res.scalars().all()) def _match_rule( rules: list[CmdbCostAllocationRule], *, provider: str, integration_id: str, resource_group: str | None, service: str | None, tags: dict[str, str], ) -> dict[str, Any]: """Walk the rule ladder, return the first match's assignment. All non-NULL match conditions on a rule are AND-ed. Returns a dict with 'person_id'/'department'/'cost_center'/'rule_id' (each may be None). Empty dict on no match. Semantics worth knowing: * ``match_integration_ids`` — None or empty list means "match any integration" (treated identically). A non-empty list restricts to that whitelist. * ``match_tag_value`` is meaningful only when paired with ``match_tag_key`` — value-without-key is rejected at the validator layer, but the matcher additionally treats it as "no tag constraint" so a stray rule that survived a manual DB edit fails open (matches every fact) rather than fails closed (matches nothing) in a way the admin wouldn't notice. * Storage type: ``match_integration_ids`` is a JSON column; SQLAlchemy returns a Python list when populated. We accept a JSON string fallback only because old hand-written migrations or pre-Pydantic API clients could have stored one — we log the parse error so the admin can fix it. """ for rule in rules: # Integration-id allowlist. None or [] = "match any". raw_ids = rule.match_integration_ids allowed: list[str] | None = None if isinstance(raw_ids, list): allowed = raw_ids if raw_ids else None # [] → None elif isinstance(raw_ids, str) and raw_ids: try: parsed = json.loads(raw_ids) except (TypeError, ValueError): logger.warning( "allocation rule %s has malformed JSON in " "match_integration_ids=%r — treating as match-any", rule.id, raw_ids[:80], ) parsed = None if isinstance(parsed, list) and parsed: allowed = parsed if allowed is not None and integration_id not in allowed: continue if ( rule.match_provider and rule.match_provider.lower() != (provider or "").lower() ): # Provider slugs in ``integrations.provider`` are # lowercase by convention (azure / aws / aliyun / # alicloud_intl / tencent / gcp / ucloud), and the # rule-creation form constrains its dropdown to the same # set. The ``.lower()`` on both sides is defence-in-depth # against rules created via direct DB edits / future # imports / hand-typed API clients picking up an # uppercase variant — those would silently never match # without this normalisation. continue # Resource-group / service / tag-value matching is case- # INSENSITIVE: cloud APIs are inconsistent about RG-name casing # — Azure RG names are case-insensitive on the cloud side # (``MyResourceGroup`` and ``myresourcegroup`` are the same # resource group), but Azure Cost Management may return either # casing depending on tenant / API version. AWS account # aliases sometimes lowercase. Pre-fix an admin who copied an # RG name out of the Azure portal as ``MyRG`` would write a # rule that silently never matched the lowercased fact emitted # by Cost Management. The same hazard applies to service names # (``Microsoft.Compute`` vs ``microsoft.compute``) and tag # values. The ``match_provider`` path already lowercases on # both sides — extend the same defence in depth here. if rule.match_resource_group: rg_norm = (resource_group or "").lower() if rule.match_resource_group.lower() != rg_norm: continue if rule.match_service: svc_norm = (service or "").lower() if rule.match_service.lower() != svc_norm: continue # Tag matching: key is the gate. Value-without-key is a # configuration error caught upstream, but be defensive. # Tag KEYS are case-sensitive on every cloud (verified Azure / # AWS / GCP) — preserve case-sensitivity on the key. Tag # VALUES are case-sensitive on Azure / AWS but case-INsensitive # on the operator's intent (admins type "team-platform" and # "Team-Platform" interchangeably); use case-insensitive # comparison on the value to match operator intuition. if rule.match_tag_key: tag_val = tags.get(rule.match_tag_key) if tag_val is None: continue # Tag values from cloud APIs occasionally arrive as # non-string types — Aliyun BSS sometimes maps ``tags.value`` # JSON booleans through unconverted, and Azure Cost # Management can emit numeric strings the JSON decoder # promotes to int. Pre-lowercase fix the ``!=`` comparison # against rule.match_tag_value (always a string) returned # non-match silently. Post-lowercase fix # ``tag_val.lower()`` raises ``AttributeError`` on non-str. # Defang via ``str(tag_val).lower()``. if ( rule.match_tag_value and rule.match_tag_value.lower() != str(tag_val).lower() ): continue # All conditions matched. return { "rule_id": str(rule.id), "person_id": ( str(rule.assign_person_id) if rule.assign_person_id else None ), "department": rule.assign_department, "cost_center": rule.assign_cost_center, } return {} async def recompute_allocation( session: AsyncSession, *, days: int = 90, ) -> dict[str, int]: """Re-evaluate the rule ladder + scope-owner fallback for the last ``days`` of facts. Used by the admin "重算分摊" button after a rule edit OR after a scope-ownership change. Walks matching facts in chunks (10k at a time) so a 90-day re-stamp on a busy tenant doesn't lock the table. Returns a summary: { "scanned": N, # facts visited in the window "matched_by_rule": M, # facts attributed via CmdbCostAllocationRule "matched_by_scope": K, # facts attributed via cmdb_cloud_scopes.owner_person_id "still_unallocated": U, # no rule, no scope owner — surfaces as 未分摊 "updated": W, # rows whose allocation columns actually changed # Backwards-compat aliases (do not remove without coordinating # with the frontend i18n key cost.alloc.recomputed): "matched": M + K, # what the dashboard shows as 命中 } Concurrency contract: this method does NOT row-lock against a racing ``ingest_cost_facts`` writer. Both writers are deterministic and evaluate the same rule ladder + scope-owner map against the same fact dimensions; the next recompute converges if they diverge transiently. ``fact_key`` UNIQUE prevents duplicate-row insertion regardless. A future PR may add SELECT FOR UPDATE if production traffic surfaces an actual collision symptom. """ # UTC for parity with ``CmdbCostFact.date`` (UTC-anchored # column). Process-local would be off-by-one for SH-anchored # cron data on a UTC-deployed worker. from datetime import UTC cutoff = datetime.now(UTC).date() - timedelta(days=days) rules = await _load_active_rules(session) # Pre-load integration → provider mapping (small). integ_rows = await session.execute( select(Integration.id, Integration.provider) .where(Integration.kind == "cloud"), ) provider_by_int = { str(r.id): str(r.provider or "") for r in integ_rows } # Walk every active scope across every cloud — recompute is a # multi-tenant operation by design (admin button restamps the # whole table, not one tenant). The set of unique owners feeds # the person-meta cache so dept/cost-center resolution is # bulk-loaded. scope_owner_lookup = await _build_scope_owner_lookup(session) bailian_key_owners = await load_api_key_owner_map(session) person_meta_cache = await _build_person_meta_cache( session, rules, extra_person_ids=( set(scope_owner_lookup.values()) | set(bailian_key_owners.values()) ) or None, ) scanned = 0 matched_by_rule = 0 # 精确到 API Key 的百炼归属,与「整账号兜底」分开计数 —— 回填后 # 这两个数的比例就是「多少钱真正落到了人头上」。 matched_by_bailian_key = 0 matched_by_scope = 0 still_unallocated = 0 # SP rows are intentionally skipped from allocation (PR #339 R2). # Counted in their own bucket so the admin "X rows still unallocated" # metric isn't polluted by ~30% of AWS rows that are correctly # left unattributed by design. sp_skipped = 0 updated = 0 batch_size = 10_000 # Never use a table-wide id cursor with ``date >= cutoff`` here. Even # an id-banded cursor needs TiDB to visit every physical row in the # band before it can reject old history, and production has large # historical id gaps. The query was consequently still exposed to a # TiDB 8176 memory-controller kill as data grew. Scan each accounting # day separately instead: ``(date, id)`` is an index range, so every # statement touches only one day's in-window facts and its own 10k-row # cursor page. ``max_date`` deliberately includes future-dated facts, # preserving the old ``date >= cutoff`` semantics. max_date = (await session.execute( select(func.max(CmdbCostFact.date)).where(CmdbCostFact.date >= cutoff), )).scalar() scan_date = cutoff # Target tuple → list of fact ids that need re-stamping to it, scoped # to the current scan batch. Grouping by the (person, dept, cc, rule, # source) the row should END UP with collapses a re-stamp of N rows # into one ``UPDATE … WHERE id IN (…)`` per distinct target instead of # N single-row round-trips. At this tenant's scale the 90-day window # holds ~150 distinct targets, so a full ~500k-row re-stamp drops from # ~500k statements (≈10 ms each ⇒ >1 h, blows the 1800 s hard timeout) # to a few hundred bulk statements (seconds). The per-row path was the # root cause of the 2026-05-27 nightly timeout: an HRIS dept/CC reorg # dirtied ~500k facts and the loop could not re-stamp them in one tick. while max_date is not None and scan_date <= max_date: last_id = 0 while True: rows = (await session.execute( select( CmdbCostFact.id, CmdbCostFact.integration_id, CmdbCostFact.scope_id, CmdbCostFact.resource_group, # 百炼阶梯要读它(API Key id 编码在首段)。这份 # SELECT 是**逐列**列举的,新阶梯用到哪列就得加哪列 # —— 漏了会在扫到第一条行时 AttributeError,而这个 # 循环是夜间 cron,炸在生产而不是 CI。 CmdbCostFact.resource_id, CmdbCostFact.service, CmdbCostFact.tags, CmdbCostFact.cost_owner_person_id, CmdbCostFact.cost_owner_department, CmdbCostFact.cost_center, CmdbCostFact.allocation_rule_id, CmdbCostFact.auto_owner_source, ) .where( CmdbCostFact.id > last_id, CmdbCostFact.date == scan_date, ) .order_by(CmdbCostFact.id.asc()) .limit(batch_size), )).all() if not rows: break # Reset per batch so each id list stays bounded by ``batch_size`` # (≤10k ids ⇒ a safe IN clause well under the placeholder limit), # which in turn bounds each commit's write set — no long-held # transaction / row-lock pile-up while re-stamping the window. pending: dict[ tuple[str | None, str | None, str | None, str | None, str], list[int], ] = defaultdict(list) for row in rows: scanned += 1 tag_dict: dict[str, str] = {} tags_raw = row.tags if isinstance(tags_raw, dict): tag_dict = {str(k): str(v) for k, v in tags_raw.items()} elif isinstance(tags_raw, str) and tags_raw: try: tag_dict = { str(k): str(v) for k, v in json.loads(tags_raw).items() } except (ValueError, TypeError): tag_dict = {} # SP discount / fee rows are never user-attributable spend; # ``ingest_cost_facts`` stamped them with # ``auto_owner_source='savings_plan'`` and NULL owner. A # rule edit / scope-owner change should NOT re-attribute # them. Same rationale as the ingest path. Round-2 review. is_sp_row = row.service == "Savings Plans" if is_sp_row: new_person = None new_department = None new_cc = None new_rule = None new_source = "savings_plan" else: allocation = _match_rule( rules, provider=provider_by_int.get( str(row.integration_id), "", ), integration_id=str(row.integration_id), resource_group=row.resource_group, service=row.service, tags=tag_dict, ) new_person = allocation.get("person_id") new_department = allocation.get("department") new_cc = allocation.get("cost_center") new_rule = allocation.get("rule_id") new_source = "rule" if new_rule else "none" # 百炼 Key 归属 —— 与 ingest 路径同一条阶梯、同一个顺序 # (规则 → Key → 订阅 Owner)。两边不一致会让「重算」把 # 摄取时算对的归属改错,而且只在回填之后才看得出来。 if not new_person and not is_sp_row: bailian_key_id = bailian_api_key_id_from_resource_id( row.service, row.resource_id, ) if bailian_key_id: bailian_owner = bailian_key_owners.get(bailian_key_id) if bailian_owner: new_person = bailian_owner new_source = AUTO_OWNER_SOURCE_BAILIAN_KEY if not new_person and row.scope_id and not is_sp_row: fallback_person = scope_owner_lookup.get(str(row.scope_id)) if fallback_person: new_person = fallback_person new_source = "scope_owner" if new_person: meta = person_meta_cache.get(new_person) or (None, None) if new_department is None: new_department = meta[0] if new_cc is None: new_cc = meta[1] if new_source == "rule": matched_by_rule += 1 elif new_source == AUTO_OWNER_SOURCE_BAILIAN_KEY: matched_by_bailian_key += 1 elif new_source == "scope_owner": matched_by_scope += 1 elif new_source == "savings_plan": sp_skipped += 1 else: still_unallocated += 1 if ( new_person != row.cost_owner_person_id or new_department != row.cost_owner_department or new_cc != row.cost_center or new_rule != row.allocation_rule_id or new_source != row.auto_owner_source ): pending[ (new_person, new_department, new_cc, new_rule, new_source) ].append(row.id) updated += 1 last_id = row.id # Flush this batch's re-stamps as one UPDATE per distinct target. for ( p_person, p_department, p_cc, p_rule, p_source, ), ids in pending.items(): await session.execute( update(CmdbCostFact) .where(CmdbCostFact.id.in_(ids)) .values( cost_owner_person_id=p_person, cost_owner_department=p_department, cost_center=p_cc, allocation_rule_id=p_rule, auto_owner_source=p_source, # SQLAlchemy core UPDATE does not trigger the ORM # ``onupdate=func.now()`` hook on ``updated_at``. Set it # explicitly so the timestamp reflects when the # allocation was last recomputed — without this the # column stays at the original ingest time and the # dashboard "last updated" display lies. updated_at=func.now(), ), ) await session.commit() scan_date += timedelta(days=1) return { "scanned": scanned, "matched_by_rule": matched_by_rule, # 精确到百炼 API Key 的归属。与 ``matched_by_scope``(整账号 # 兜底)分开报:回填后这两个数的比例就是「多少笔真正落到人头上」, # 合成一个数就看不出这次改动到底有没有生效。 "matched_by_bailian_key": matched_by_bailian_key, "matched_by_scope": matched_by_scope, "still_unallocated": still_unallocated, # SP discount / fee rows intentionally not attributed to a # person; tracked separately so the admin's "X unallocated" # tile isn't polluted by ~30% of AWS rows. PR #339 R2. "sp_skipped": sp_skipped, "updated": updated, # Backwards-compat: existing frontend i18n key # ``cost.alloc.recomputed`` reads ``matched`` to render # "命中 N 条". Keep the meaning aligned — total successful # attributions, regardless of source (百炼 Key 归属也是一次成功 # 归属,必须计入,否则前端的「命中 N 条」会比实际少)。 "matched": ( matched_by_rule + matched_by_bailian_key + matched_by_scope ), } # ─── Analytics ────────────────────────────────────────────────────── # Group-by alias → SQL column expression. The router validates against # this set; anything not listed here returns 400. GROUP_BY_COLUMNS: dict[str, Any] = { "service": CmdbCostFact.service, "resource_group": CmdbCostFact.resource_group, "resource_id": CmdbCostFact.resource_id, "region": CmdbCostFact.region, "resource_type": CmdbCostFact.resource_type, "cost_center": CmdbCostFact.cost_center, "department": CmdbCostFact.cost_owner_department, "owner": CmdbCostFact.cost_owner_person_id, "integration": CmdbCostFact.integration_id, } def _month_chunks(start: date, end: date) -> list[tuple[date, date]]: """Split an inclusive ``[start, end]`` window into per-calendar-month inclusive sub-windows (clipped to the requested bounds). TiDB 8175 防线（项目规则，非可选）：``cmdb_cost_facts`` 上的宽 GROUP BY 聚合必须按月分块 —— 2026-07-28 生产事故里，财务瀑布一次 GROUP BY 扫 6 个月（~550 万行）打爆 TiDB serverless 的单 SQL 内存 配额（错误 8175）。``breakdown`` / ``timeseries`` 接受最长 365 天 的窗口（~2 倍于事故行数），且迁移 431 的覆盖索引不含它们的分组 键，同样必须收窄单条 SQL 的扫描窗口；分块结果在 Python 侧全量 合并（top-N 截断只发生在合并之后），与未分块聚合逐键一致。 """ chunks: list[tuple[date, date]] = [] cur = start while cur <= end: if cur.month == 12: next_month = date(cur.year + 1, 1, 1) else: next_month = date(cur.year, cur.month + 1, 1) chunk_end = min(end, next_month - timedelta(days=1)) chunks.append((cur, chunk_end)) cur = chunk_end + timedelta(days=1) return chunks def _grouped_sort_key(item: tuple[Any, Decimal]) -> tuple[Decimal, str]: """Deterministic top-N ordering for chunk-merged grouped totals: cost desc, then key (NULL first, stringified) as the tiebreaker.""" key, cost = item return (-cost, "" if key is None else str(key)) def _dec(value: Any) -> Decimal: """DB aggregate → ``Decimal``, NULL → 0. Money is accumulated in ``Decimal`` and converted to float exactly once, at the response boundary (project rule). Month-chunking turned each group total from ONE DB-side ``SUM`` into up to 12 Python additions — in float those additions accumulate representation error inside the cent the dashboard prints. """ if value is None: return Decimal(0) if isinstance(value, Decimal): return value return Decimal(str(value)) # 分块间并发度。与财务瀑布同值(cost_router 的 ``Semaphore(3)``):实测 # TiDB serverless 上并发抢 RU 会拖慢单块,3 是提速与互踩之间的平衡点。 _CHUNK_CONCURRENCY = 3 # ``top_resources`` merges a per-chunk top slice rather than every group # (resource_id is the highest-cardinality dimension in the table, so a # year of full groups would just move the memory problem into Python). # The slice is this multiple of the caller's ``limit``, floored at 2000: # a resource has to rank below ~2000 in EVERY month yet inside the top-N # for the year to be missed, which flat spend cannot do. A chunk that # reaches the cap logs — silent truncation would read as a complete # "top resources" list. _RESOURCE_CHUNK_OVERSAMPLE = 20 async def _run_month_chunks( session: AsyncSession, chunks: list[tuple[date, date]], runner: Callable[[AsyncSession, date, date], Awaitable[Any]], session_factory: Any | None, ) -> list[Any]: """Run ``runner`` once per month chunk; results in chunk order. Same execution strategy as the finance waterfall (see ``cost_router.finance_waterfall_endpoint``): with a ``session_factory`` the chunks fan out under a :data:`_CHUNK_CONCURRENCY`-wide semaphore, EACH on its own session (one ``AsyncSession`` cannot serve concurrent statements). The HTTP layer passes ``app.core.database.async_session_factory``; a 365-day window is 12 chunks × N aggregates, and sequentially that is ~12× the latency floor of the pre-chunking single query. Without a factory the chunks run sequentially on the caller's session. That is the required behaviour for every non-HTTP caller (cron / MCP / tests / any code holding uncommitted rows): a separate session lives in its own transaction and cannot see the caller's uncommitted state. ``runner`` must return plain values (no ORM instances), because its session is closed before the result is merged. """ if session_factory is None: return [await runner(session, cs, ce) for cs, ce in chunks] sem = asyncio.Semaphore(_CHUNK_CONCURRENCY) async def _one(cs: date, ce: date) -> Any: async with sem: async with session_factory() as chunk_session: return await runner(chunk_session, cs, ce) return list(await asyncio.gather(*[_one(cs, ce) for cs, ce in chunks])) async def _department_filter_clause( session: AsyncSession, department: str | None, ) -> Any | None: """Resolve a department-name filter into a WHERE clause over the denormalised ``cmdb_cost_facts.cost_owner_department`` column. Returns ``None`` when no department filter is requested (caller skips filtering) and an ``IN (...)`` clause otherwise. Picking a parent department rolls up its whole subtree (:func:`app.core.org_scope.resolve_department_subtree_names`) so the billing-tab cost ranking narrows to the SAME set of people the cloud- overview headcount / subscription / IAM filters use — a parent pick means "this department and everything under it" everywhere. Facts with a NULL ``cost_owner_department`` (unattributed spend — e.g. Savings-Plan netting rows, or scopes with no owner) are intentionally EXCLUDED under the filter: that spend is not attributable to the chosen department, so the filtered total is a true subset of the fleet total rather than silently folding unallocated cost into the picked department. """ names = await resolve_department_subtree_names(session, department) if names is None: return None return CmdbCostFact.cost_owner_department.in_(names) async def _scope_filter_clauses( session: AsyncSession, scope_id: str | None, ) -> list[Any]: """Resolve a subscription/scope filter into WHERE clauses over ``cmdb_cost_facts``. Reuses the SAME routing as the per-scope billing drawer (:func:`scope_billing_detail.scope_fact_filters`): a tenant-level scope (Azure subscription / AWS account / GCP project / …) filters on ``scope_id``; an RG-typed scope filters on ``(integration_id, resource_group)``. Threading this into :func:`breakdown` / :func:`timeseries` lets the cloud-overview drilldown isolate ONE subscription (e.g. the Azure OpenAI sub) and group its resources — the "per-Foundry" view path. Because both surfaces share ``scope_fact_filters``, the drilldown total for a scope agrees to the cent with that scope's billing tab. Returns ``[]`` when no filter is requested. Raises ``ValueError`` on an unknown scope_id (the router maps it to a 400). When the scope resolves but yields no usable routing (an RG-typed scope missing its ``external_id``), returns a guaranteed-false clause so the drilldown shows an honest empty set rather than silently widening to the whole fleet. """ if not scope_id: return [] # Lazy import — scope_billing_detail imports models/identity only, so # there is no top-level cycle, but keep it local to be defensive # against future import-order churn. from app.domains.cmdb.scope_billing_detail import scope_fact_filters scope = (await session.execute( select(CmdbCloudScope).where(CmdbCloudScope.id == scope_id), )).scalar_one_or_none() if scope is None: raise ValueError(f"unknown scope_id={scope_id!r}") clauses, _routing = scope_fact_filters(scope) if not clauses: return [false()] return clauses async def _resolve_owner_labels( session: AsyncSession, person_ids: list[str], ) -> dict[str, str]: """Map ``cost_owner_person_id`` → a human-readable label. The ``owner`` group_by dimension stores the bare ``cost_owner_person_id`` UUID in ``key``/``label``. Surfacing a raw UUID in the "归属人" drilldown is useless to finance — "这个钱是哪些 人花的" needs names. Resolve the UUIDs to ``display_name`` (falling back to ``work_email``, then the raw id for a since-deleted person) so the pie / timeseries / table all read as people, not opaque ids. Returns ``{str(person_id): label}``; ids that no longer resolve are simply absent and the caller keeps the original key as the label. """ ids = [p for p in person_ids if p] if not ids: return {} rows = (await session.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ).where(IdentityPerson.id.in_(ids)), )).all() return { str(r.id): (r.display_name or r.work_email or str(r.id)) for r in rows } async def monthly_measure_rollup( session: AsyncSession, *, month_start: date, month_end: date, integration_id: str | None = None, ) -> list[Any]: """单个自然月的双口径聚合（成本口径改造 PR7，8175 修复）。 财务瀑布与「成本口径」对数区的共享查询单元。**窗口刻意收窄到 一个月**：2026-07-28 生产事故里，瀑布一次 GROUP BY 扫 6 个月 facts（~550 万行）打爆 TiDB serverless 的单 SQL 内存配额（错误 8175）。按月分块从根上封顶单查询内存；配合覆盖索引 ``ix_facts_date_measure``（迁移 431）每月聚合走 IndexRangeScan 不回表。 刻意不 JOIN integrations：provider 由调用方用一次小表查询映射 （见 :func:`integration_provider_map`）——JOIN 会把执行计划推回 宽行 HashJoin，破坏覆盖索引的意义。 返回行：``(integration_id, cost_bucket, invoice, list_sum, public_sum, savings_sum, paired_public_sum, paired_list_sum)``。 ``savings_sum`` = Σ(public − list) **仅对同时携带两值的行**。 「研发自主节省」是 public 与 list 的逐行配对差；SQL 裸 ``SUM(public) − SUM(list)`` 的两个 SUM 覆盖面不同（AWS org BQ 行 带 public，AWS SG CE 行 public ≡ NULL 但 list 有值），相减会把 无 public 覆盖的整段 list 支出当成负节省。CASE 过滤保证差值只在 public 覆盖的行上累计。 ``paired_public_sum`` / ``paired_list_sum``（2026-07-29）是**同一 配对子集**（同一 CASE 条件）上的两个分量和，节省项因此有闭合算术： ``paired_public_sum − paired_list_sum ≡ savings_sum``。没有它们， 调用方只能把覆盖面更宽的 ``public_sum`` / ``list_sum`` 与节省项并排 展示 —— 财务读到的一行三个数加不起来（复审实证：public 100000、 节省 10000、list 110000）。``list_sum`` 保持全组口径： ``contract_inprice_discount``（list − usage 桶发票额）与「成本口径」 对数区都按它计价，收窄会把只有 list 覆盖的整段支出（AWS SG CE）从 折前成本里抹掉。 """ from sqlalchemy import case where = [ CmdbCostFact.date >= month_start, CmdbCostFact.date < month_end, ] if integration_id: where.append(CmdbCostFact.integration_id == integration_id) # 配对覆盖:两个分量都非 NULL 的行 —— 节省项与它的两个基数共用这一 # 个条件,三者的算术才闭合。 paired = and_( CmdbCostFact.amount_usd_public.is_not(None), CmdbCostFact.amount_usd_list.is_not(None), ) return list((await session.execute( select( CmdbCostFact.integration_id, CmdbCostFact.cost_bucket, func.coalesce(func.sum(CmdbCostFact.amount_usd), 0).label( "invoice", ), func.sum(CmdbCostFact.amount_usd_list).label("list_sum"), func.sum(CmdbCostFact.amount_usd_public).label("public_sum"), func.sum( case( ( paired, CmdbCostFact.amount_usd_public - CmdbCostFact.amount_usd_list, ), else_=0, ), ).label("savings_sum"), # 配对子集的两个基数。SUM(CASE … ELSE NULL) 而非 ELSE 0: # 整组无配对行时必须是 NULL(口径未接入),不能是 0(一个 # 关于牌价的断言)。 func.sum( case((paired, CmdbCostFact.amount_usd_public)), ).label("paired_public_sum"), func.sum( case((paired, CmdbCostFact.amount_usd_list)), ).label("paired_list_sum"), ) .where(and_(*where)) .group_by(CmdbCostFact.integration_id, CmdbCostFact.cost_bucket), )).all()) async def excluded_bucket_summary( session: AsyncSession, *, start: date, end: date, integration_id: str | None = None, scope_id: str | None = None, department: str | None = None, include_amounts: bool = False, session_factory: Any | None = None, ) -> dict[str, Any]: """研发口径行白名单**排除掉**的行 —— 按桶给出笔数（+ 可选金额）。 折前口径只统计 ``cost_bucket='usage'``（见 :func:`app.domains.cmdb.cost_view.fact_engineering_amount`）。退款、 承诺费、折扣、赠金、税、Support 都不进总额 —— 正确，但如果读者 在页面上完全看不到它们，一笔 −$13,229.89 的集中退款就成了「凭空 消失的钱」，比之前把它算进总额更难解释。这个函数就是那条可见性 通道：**同一个筛选条件**（时间窗 / 集成 / 订阅 / 部门）下，被排除 的行有多少笔、分别属于哪个桶。 ``include_amounts`` 是保密红线的执行点，由调用方按 :func:`app.domains.cmdb.cost_view.can_view_invoice_measure` 传入， **默认 False**： * ``False``（研发可见面）→ 每桶只回 ``count``，``invoice_usd`` 为 None。金额留空不是偷懒 —— 折前总额与非 usage 桶的实付额同框， 两者一相除就能反推公司折扣率，正是 cost_view 模块开头那条红线 要挡的事。 * ``True``（finance / cloud_admin / global_admin）→ 附带该桶的实付 合计。这些角色本来就能请求 ``measure=invoice``，不构成新增泄露。 金额一律实付口径（``amount_usd``）：非 usage 行的 ``amount_usd_list`` 按建模契约恒 NULL，"折前退款" 不是一个有意义的量。 按月分块（TiDB 8175 防线，项目规则）；``session_factory`` 语义同 :func:`_run_month_chunks`。 :func:`breakdown` **不调用本函数** —— 它把 :func:`_excluded_bucket_rows` 直接搭在自己那一轮分块里执行。本函数 是给独立调用方（订阅账单页等）用的入口，两边共用同一条 SQL 与同一 个合并/呈现助手，不存在两份口径。 """ base_clauses: list[Any] = [non_engineering_row_clause()] if integration_id: base_clauses.append(CmdbCostFact.integration_id == integration_id) base_clauses.extend(await _scope_filter_clauses(session, scope_id)) dept_clause = await _department_filter_clause(session, department) if dept_clause is not None: base_clauses.append(dept_clause) async def _chunk( chunk_session: AsyncSession, chunk_start: date, chunk_end: date, ) -> list[tuple[str, int, Decimal]]: return await _excluded_bucket_rows( chunk_session, [ CmdbCostFact.date >= chunk_start, CmdbCostFact.date <= chunk_end, *base_clauses, ], ) merged: dict[str, tuple[int, Decimal]] = {} for chunk_rows in await _run_month_chunks( session, _month_chunks(start, end), _chunk, session_factory, ): _merge_excluded_rows(merged, chunk_rows) return _excluded_payload(merged, include_amounts=include_amounts) async def _excluded_bucket_rows( session: AsyncSession, clauses: list[Any], ) -> list[tuple[str, int, Decimal]]: """一个分块内、被研发口径排除的行：``(bucket, 笔数, 实付合计)``。 ``clauses`` 必须已经包含 :func:`non_engineering_row_clause` 与本次 读的时间窗 / 筛选条件。 """ rows = (await session.execute( select( CmdbCostFact.cost_bucket, func.count(CmdbCostFact.id).label("n"), func.coalesce(func.sum(CmdbCostFact.amount_usd), 0).label( "invoice", ), ) .where(and_(*clauses)) .group_by(CmdbCostFact.cost_bucket), )).all() return [ (str(r.cost_bucket), int(r.n or 0), _dec(r.invoice)) for r in rows ] def _merge_excluded_rows( merged: dict[str, tuple[int, Decimal]], rows: list[tuple[str, int, Decimal]], ) -> None: for bucket, n, invoice in rows: prev_n, prev_amt = merged.get(bucket, (0, Decimal(0))) merged[bucket] = (prev_n + n, prev_amt + invoice) def _excluded_payload( merged: dict[str, tuple[int, Decimal]], *, include_amounts: bool, ) -> dict[str, Any]: buckets = [ { "bucket": bucket, "count": n, "invoice_usd": ( float(round(amount, 2)) if include_amounts else None ), } # 笔数降序:读者最先看到「量最大的那类被排除行」。 for bucket, (n, amount) in sorted( merged.items(), key=lambda kv: (-kv[1][0], kv[0]), ) ] return { "buckets": buckets, "total_count": sum(b["count"] for b in buckets), "amounts_visible": include_amounts, } async def integration_provider_map(session: AsyncSession) -> dict[str, str]: """integration_id → provider（integrations 是小表，一次全量）。""" rows = (await session.execute( select(Integration.id, Integration.provider), )).all() return {str(r.id): str(r.provider) for r in rows} async def breakdown( session: AsyncSession, *, start: date, end: date, group_by: str, integration_id: str | None = None, scope_id: str | None = None, department: str | None = None, top: int = 20, measure: str = "list", include_excluded_amounts: bool = False, session_factory: Any | None = None, ) -> dict[str, Any]: """Aggregate cost over a date range, grouped by one dimension. ``session_factory`` (optional) hands the month-chunk fan-out its own sessions so the chunks can run with bounded concurrency — see :func:`_run_month_chunks`. The HTTP layer passes ``async_session_factory``; every other caller leaves it None and the chunks run sequentially on ``session``. ``include_excluded_amounts`` gates the **amounts** inside ``excluded_buckets`` (see :func:`excluded_bucket_summary`) — the caller passes ``can_view_invoice_measure(role_names)``. It varies the response body by role, so the HTTP layer MUST also fold it into the Redis cache key or one role eats another's payload. Contract — IMPORTANT to keep the dashboard math honest: * ``total_usd`` is the canonical sum: ``sum(groups) + other_usd``. * ``excluded_buckets`` is **outside** that sum by design: those rows (退款 / 承诺费 / 折扣 / 赠金 / 税 / Support) are not engineering spend. It exists so a large refund is visibly excluded rather than silently missing. ``None`` under ``measure=invoice``. * ``unallocated_usd`` is **informational** — a SUBSET of ``total_usd``, NOT an additional bucket. It's the slice of ``total_usd`` whose underlying facts have NO cost owner (``cost_owner_person_id IS NULL``). When ``group_by`` is one of the allocation-output columns (cost_center / department / owner), the NULL group inside ``groups`` IS the unallocated slice — labelled ``"未分摊"`` so the dashboard reader doesn't mistake it for "facts with empty-string cost-center". Why this column and not ``allocation_rule_id IS NULL`` (the pre-PR-241 definition): PR #241 added a scope-owner fallback to ``ingest_cost_facts`` and ``recompute_allocation`` so a fact whose rule ladder misses still attributes to ``cmdb_cloud_scopes.owner_person_id``. The fallback writes ``cost_owner_person_id`` (and downstream ``cost_owner_department`` / ``cost_center``) but does NOT write ``allocation_rule_id`` because the match path is scope-based, not rule-based. Using ``allocation_rule_id IS NULL`` as the unallocated signal therefore counts every scope-owner-attributed fact as unallocated — that's the 2026-05-11 prod symptom where the dashboard showed $7,316.10 / 99.8% unallocated even though ``recompute_allocation`` summary returned ``matched_by_scope=10498, still_unallocated=0``. The two readers disagreed because the KPI's filter still used the rule-only definition. The canonical "who pays this cost" signal is ``cost_owner_person_id``, regardless of which match path populated it. SP discount / fee rows (``auto_owner_source='savings_plan'``) are NULL-owner by design but are EXCLUDED from this scalar and from the NULL/"未分摊" group — they are surfaced as their own pinned ``"Savings Plans"`` group (``is_discount``) so the KPI tile and the chart's unallocated slice agree, and neither is polluted by the org-level SP netting. The pinned group carries ``is_unallocated=False``; ``total_usd`` still nets the SP amount in (gross-vs-net reconciliation). * ``other_usd`` collapses everything past the top-N rank. Frontend MUST NOT add ``unallocated_usd`` to ``total_usd`` to get a "grand total" — that would double-count. """ # 口径路由(权限判定在 router 层完成;这里只选列)。研发口径 = # COALESCE(list, invoice) 过渡语义,见 cost_view 模块 docstring。 amount_col = fact_amount_expr(measure) is_alloc_dim = group_by in {"cost_center", "department", "owner"} # ``provider`` is the one dimension that does NOT live on # ``cmdb_cost_facts`` — it's ``integrations.provider``. Group on the # joined column (the ``Integration`` join is added to the main grouped # query below; the SP / unallocated / unconverted sub-queries filter # only on cost_facts columns so they need no join). The key/label is # the bare provider string — no UUID resolution like ``owner`` needs. is_provider_dim = group_by == "provider" if is_provider_dim: col = Integration.provider else: col = GROUP_BY_COLUMNS.get(group_by) if col is None: raise ValueError( f"unknown group_by={group_by!r}; supported: " f"{sorted([*GROUP_BY_COLUMNS, 'provider'])}", ) base_clauses = [] if integration_id: base_clauses.append(CmdbCostFact.integration_id == integration_id) base_clauses.extend(await _scope_filter_clauses(session, scope_id)) dept_clause = await _department_filter_clause(session, department) if dept_clause is not None: base_clauses.append(dept_clause) # PR #339 R2 / Round-XX SP-pin generalisation: Savings Plans (SP) # discount & fee rows are org-level commitment netting, NOT # user-attributable spend. ``ingest_cost_facts`` / # ``recompute_allocation`` stamp them with # ``auto_owner_source='savings_plan'`` and a NULL cost-owner. On the # allocation dimensions (owner / department / cost_center) they fold # into the NULL "未分摊" bucket; on ``service`` they form a "Savings # Plans" key row. Either way they distort whatever bucket they land # in — the unallocated KPI tile and the chart's 未分摊 slice would # BOTH carry the (usually large negative) SP amount, and finance # can't reconcile gross vs net. # # Fix (was service-only, now every ``group_by``): EXCLUDE SP rows # from the main grouped query AND the unallocated scalar, sum them # once via a dedicated NULL-safe query, and re-add them as a single # pinned group (``is_unallocated=False``, ``is_discount=<0``). # ``total_usd`` is unchanged — it nets the SP discount back in # (gross-vs-net reconciliation) by adding ``sp_sum`` to the non-SP # grouped total. This also makes the top-N edge case trivial: SP is # never inside ``other_usd`` or the NULL bucket regardless of rank. # # NULL-safe identification: a row is SP iff # ``auto_owner_source = 'savings_plan'``. A NULL ``auto_owner_source`` # (pre-column / un-recomputed legacy rows) is treated as NOT SP, so # the ``coalesce(..., '')`` guard keeps those rows in their natural # bucket rather than silently dropping them out of the breakdown. not_sp = func.coalesce(CmdbCostFact.auto_owner_source, "") != "savings_plan" # TiDB 8175 防线:四个聚合全部按月分块(``_month_chunks``,项目 # 规则,非可选)——单条 SQL 的日期窗口 ≤1 个自然月。分块结果在 # Python 侧全量合并;top-N 截断只发生在合并之后,所以跨月分组键 # 与「other」桶的数学与未分块一致。分块间的执行策略(有界并发 + # 独立 session,或顺序执行)见 :func:`_run_month_chunks`;金额一律 # Decimal 累加,只在响应边界转 float。 async def _chunk( chunk_session: AsyncSession, chunk_start: date, chunk_end: date, ) -> tuple[ list[tuple[Any, Decimal]], Decimal, int, Decimal, # (currency, row_count, native_total) —— 未折算 USD 的原币行 list[tuple[Any, int, Decimal]], # (cost_bucket, row_count, invoice_total) —— 被研发口径排除的行 list[tuple[str, int, Decimal]], ]: chunk_clauses = [ CmdbCostFact.date >= chunk_start, CmdbCostFact.date <= chunk_end, *base_clauses, ] grouped_stmt = select( col.label("key"), func.coalesce(func.sum(amount_col), 0).label("cost"), ) if is_provider_dim: grouped_stmt = grouped_stmt.select_from(CmdbCostFact).join( Integration, Integration.id == CmdbCostFact.integration_id, ) grouped_stmt = ( grouped_stmt .where(and_(*chunk_clauses, not_sp)) .group_by(col) ) grouped = [ (r.key, _dec(r.cost)) for r in (await chunk_session.execute(grouped_stmt)).all() ] # Dedicated SP sum + count. Independent of ``group_by`` so the # pin works identically for service / owner / department / # cost_center / region / … . ``sp_count`` lets us skip the # pinned group entirely for tenants with no SP activity (e.g. # GCP-only) instead of rendering a spurious "$0.00 Savings # Plans" row. sp_stat = (await chunk_session.execute( select( func.coalesce(func.sum(amount_col), 0).label("sp_sum"), func.count(CmdbCostFact.id).label("sp_count"), ).where( and_( *chunk_clauses, CmdbCostFact.auto_owner_source == "savings_plan", ), ), )).one() # Unallocated = facts where the canonical cost-owner column is # NULL, EXCLUDING SP rows (which are NULL-owner by design but # are surfaced as their own pinned group, not as unallocated). # See the docstring (above this function) for the # cost_owner_person_id rationale — pre-PR-241 used # ``allocation_rule_id IS NULL`` here, which mis-counted # scope-owner-attributed facts as unallocated. Excluding SP # keeps this scalar KPI in agreement with the NULL/"未分摊" # group in ``groups`` below and with ``recompute_allocation``'s # ``sp_skipped`` bucket. unalloc_res = await chunk_session.execute( select(func.coalesce(func.sum(amount_col), 0)) .where( and_( *chunk_clauses, CmdbCostFact.cost_owner_person_id.is_(None), not_sp, ), ), ) # Round-22 audit: rows whose currency != USD but amount_usd == 0 # are "unconverted natives" — a driver couldn't convert (missing # FX rate at ingest time, or the driver chose to defer # conversion to the analytics layer). They are SILENTLY counted # as $0 in ``total_usd`` above, so a tenant with heavy # CNY/JPY/BRL spend whose ingest pipeline lost its FX rate # window shows a USD total that materially understates real # spend. Surface a separate count + native-sum block so the FE # can render a "X 行未换算" banner. unconverted_res = await chunk_session.execute( select( func.coalesce(func.count(CmdbCostFact.id), 0).label("cnt"), func.coalesce( func.sum(CmdbCostFact.amount_native), 0, ).label("nat"), CmdbCostFact.currency.label("ccy"), ) .where( and_( *chunk_clauses, CmdbCostFact.currency != "USD", CmdbCostFact.amount_usd == 0, CmdbCostFact.amount_native != 0, ), ) .group_by(CmdbCostFact.currency) ) # 被研发口径排除的行,搭在**同一轮分块**里查(不另起一轮 # ``_run_month_chunks``)。这个端点的连接预算是精算过的:准入闸 # 按「1 条请求 session + 最多 3 条分块 session」记账,再开一轮扇出 # 会让每个请求的峰值连接翻倍,而连接池抽干过一次(群发持锁把登录 # 一起锁死,见 cost_router 的准入闸注释)。实付口径不需要这个块。 excluded_rows = ( [] if measure == COST_MEASURE_INVOICE else await _excluded_bucket_rows( chunk_session, [*chunk_clauses, non_engineering_row_clause()], ) ) return ( grouped, _dec(sp_stat.sp_sum), int(sp_stat.sp_count or 0), _dec(unalloc_res.scalar()), [ (r.ccy or "", int(r.cnt or 0), _dec(r.nat)) for r in unconverted_res.all() ], excluded_rows, ) key_totals: dict[Any, Decimal] = defaultdict(Decimal) sp_sum = Decimal(0) sp_count = 0 unallocated = Decimal(0) unconverted_acc: dict[str, dict[str, Any]] = {} excluded_acc: dict[str, tuple[int, Decimal]] = {} for ( grouped, chunk_sp_sum, chunk_sp_count, chunk_unalloc, unconverted, chunk_excluded, ) in await _run_month_chunks( session, _month_chunks(start, end), _chunk, session_factory, ): for key, cost in grouped: key_totals[key] += cost sp_sum += chunk_sp_sum sp_count += chunk_sp_count unallocated += chunk_unalloc for ccy, cnt, nat in unconverted: acc = unconverted_acc.setdefault( ccy, {"cnt": 0, "nat": Decimal(0)}, ) acc["cnt"] += cnt acc["nat"] += nat _merge_excluded_rows(excluded_acc, chunk_excluded) ordered = sorted(key_totals.items(), key=_grouped_sort_key) # 研发口径下,一个恒 0 的分组只可能由「全是被白名单排除的行」构成 # (``fact_engineering_amount`` 对非 usage 桶返回 0)—— 典型如只有 # SP 抵扣行的 "Savings Plans"、只有退款行的某个服务。把它们留在 # groups 里就是给每张图表挂一排 "$0.00" 空行,读者还会以为那是 # 「这个服务这段时间没花钱」。它们的去向由 ``excluded_buckets`` # 统一交代,不是静默丢弃。 # # 只在 list 口径下裁:实付口径统计所有桶,一个净额恰好为 0 的分组 # (用量 +5 / 退款 −5)是真实存在的服务,不能抹掉。 if measure != COST_MEASURE_INVOICE: ordered = [(k, cost) for k, cost in ordered if cost != 0] non_sp_total = sum((cost for _k, cost in ordered), Decimal(0)) # ``total_usd`` stays canonical: non-SP grouped total + SP net. This # equals the old "sum of every row" total, so the dashboard grand # total is unchanged by the pin (gross-vs-net reconciliation holds). total = non_sp_total + sp_sum unconverted_by_currency = [ { "currency": ccy, "row_count": int(acc["cnt"]), "native_total": float(acc["nat"]), } for ccy, acc in sorted(unconverted_acc.items()) ] # SP is already excluded from ``ordered`` (and therefore from both # the top-N slice and ``other_usd``); it is re-added below as a # pinned group, so the top-N math here only ever sees real # per-dimension spend. top_rows = ordered[:top] other = sum((cost for _k, cost in ordered[top:]), Decimal(0)) # Backend returns the raw key + the ``is_unallocated`` / # ``is_discount`` flags. Frontend i18ns the empty/未分摊 label so the # same payload serves zh-CN / en-US / ja-JP clients without the # backend carrying user-facing strings. groups = [ { "key": key if key is not None else "", "label": key if key else "", "cost_usd": float(round(cost, 2)), "share_pct": float(round( 100 * cost / total, 2, )) if total > 0 else 0.0, "is_unallocated": ( is_alloc_dim and (key is None or key == "") ), "is_discount": False, } for key, cost in top_rows ] # Pin the SP discount / fee as its own group for EVERY dimension. # ``is_unallocated=False`` — it is an org-level netting, not a # missing-owner fact — and ``is_discount`` drives the FE's # green/italic affordance when the net is negative (the common AWS # linked-account case). Skipped entirely when the tenant has no SP # rows in the window. # # ``sp_sum != 0`` (2026-08-15, 口径白名单修复的联动): SP 行落在 # ``commitment_fee`` / ``commitment_discount`` 桶,研发口径下 # ``fact_engineering_amount`` 对它们恒 0(公司层面项不进折前成本, # 见 cost_view 模块 docstring)。只判 ``sp_count > 0`` 会给每个有 SP # 的租户钉一条 "$0.00 Savings Plans" 空行 —— 正是这段注释一开始就想 # 避免的那种伪行。实付口径下 sp_sum 仍是真实净额,行照常出现。 if sp_count > 0 and sp_sum != 0: groups.append({ "key": "Savings Plans", "label": "Savings Plans", "cost_usd": float(round(sp_sum, 2)), "share_pct": ( float(round(100 * sp_sum / total, 2)) if total > 0 else 0.0 ), "is_unallocated": False, "is_discount": sp_sum < 0, }) # ``owner`` keys are raw person UUIDs — resolve to display names so # the drilldown answers "这个钱是哪些人花的" with people, not opaque # ids. ``key`` stays the UUID (stable rowKey); only ``label`` flips. # 研发口径下被行白名单排除的行 —— 与 groups 用同一套筛选条件, # 谓词是 groups 那条的严格补集(``non_engineering_row_clause``), # 所以「计入总额」与「解释给读者听」两边合起来正好是全集。 # 实付口径统计所有桶,不存在被排除的行 → None(前端不渲染)。 excluded = ( None if measure == COST_MEASURE_INVOICE else _excluded_payload( excluded_acc, include_amounts=include_excluded_amounts, ) ) if group_by == "owner": # Resolve only real person-UUID groups: the unallocated bucket # (key="") and the pinned "Savings Plans" discount row # (``is_discount``, a literal label, not a person) are skipped — # passing "Savings Plans" to an ``id IN (...)`` would raise on # UUID coercion. def _is_person_group(g: dict) -> bool: return bool( g["key"] and not g["is_unallocated"] and not g.get("is_discount") ) owner_labels = await _resolve_owner_labels( session, [str(g["key"]) for g in groups if _is_person_group(g)], ) for g in groups: if _is_person_group(g): g["label"] = owner_labels.get(str(g["key"]), str(g["key"])) return { "groups": groups, "other_usd": float(round(other, 2)), "total_usd": float(round(total, 2)), "unallocated_usd": float(round(unallocated, 2)), "unallocated_pct": ( float(round(100 * unallocated / total, 2)) if total > 0 else 0.0 ), # Round-22 audit: surface unconverted (FX-missing) native- # currency rows so the FE can show a "X 行未折算 USD" warning # banner. Empty list = clean totals; non-empty = total_usd is # an under-count and the operator needs to fix FX ingestion. "unconverted_by_currency": unconverted_by_currency, # 研发口径行白名单排除掉的行(退款/承诺费/折扣/赠金/税/Support)。 # 实付口径本就统计全部桶,没有「被排除」这回事 → None,前端据此 # 完全不渲染该区块。见 :func:`excluded_bucket_summary`。 "excluded_buckets": excluded, } async def subordinates_cost_breakdown( session: AsyncSession, *, person_ids: set[uuid.UUID] | set[str], start: date, end: date, integration_id: str | None = None, measure: str = "list", ) -> dict[str, Any]: """Manager-scoped cost breakdown: self + every direct/transitive report. Used by ``GET /api/v1/cmdb/cost/subordinates`` (and the matching MCP tool) so a BU lead / line manager can see "what is my org spending on cloud?" without IT-admin privilege. The caller has already resolved ``person_ids`` via the asset-domain BFS helper (:func:`app.domains.asset.router._resolve_subordinate_ids`) so this function is purely the aggregation layer. Returns four facets in one round-trip: * ``total_usd`` — grand total across the population. * ``by_person`` — one row per person with cost > 0, sorted DESC. Only persons in the input set are returned, even when their department field on the fact rows is ``NULL`` (i.e. unallocated). ``display_name`` / ``work_email`` / ``department`` / ``cost_center`` are joined from :class:`IdentityPerson`. * ``by_service`` — top services across the population (top-20). * ``by_department`` — department rollup using the denormalised ``cost_owner_department`` column on ``cmdb_cost_facts``. May include ``NULL`` rows when allocation rules haven't been set up for some line items — surfaced as ``"未分摊"`` via the ``is_unallocated`` flag the frontend i18ns. Empty result shape (``person_ids`` is empty OR no facts in window): all four facets return zero / empty list. The endpoint layer treats this as a 200 OK with the empty body — a manager with no current reports OR a fresh tenant with no ingest yet sees an empty page, not an error. The manager themselves is INCLUDED in ``person_ids`` by the caller so their own spend lands in ``by_person`` alongside their reports — consistent with the asset domain's "manager scope = self + subordinates" convention. """ # 口径路由(权限判定在 router 层完成;这里只选列)。研发口径 = # COALESCE(list, invoice) 过渡语义,见 cost_view 模块 docstring。 amount_col = fact_amount_expr(measure) if not person_ids: return { "total_usd": 0.0, "by_person": [], "by_service": [], "by_department": [], "person_count": 0, "fact_count": 0, } str_ids = [str(pid) for pid in person_ids] where_clauses = [ CmdbCostFact.cost_owner_person_id.in_(str_ids), ] if integration_id: where_clauses.append(CmdbCostFact.integration_id == integration_id) # TiDB 8175 防线(``_month_chunks``,项目规则,非可选):本函数的窗口 # 也可达 365 天,四个聚合与 ``breakdown`` 同表同宽行。按月分块 + # Python 侧合并;金额一律 Decimal 累加,只在响应边界转 float。 # ``by_service`` 的 top-20 截断挪到合并之后 —— 在分块内 LIMIT 会让 # 某个单月冠军挤掉全窗口第一(与 ``breakdown`` 的 top-N 同一道理)。 chunks = _month_chunks(start, end) def _chunk_where(chunk_start: date, chunk_end: date) -> list[Any]: return [ CmdbCostFact.date >= chunk_start, CmdbCostFact.date <= chunk_end, *where_clauses, ] # ── Total + fact count (one round-trip per chunk) ───────────── total_dec = Decimal(0) fact_count = 0 for chunk_start, chunk_end in chunks: totals_row = (await session.execute( select( func.coalesce(func.sum(amount_col), 0).label("total"), func.count(CmdbCostFact.id).label("fact_count"), ).where(and_(*_chunk_where(chunk_start, chunk_end))), )).one() total_dec += _dec(totals_row.total) fact_count += int(totals_row.fact_count or 0) # Only short-circuit on TRULY empty windows (no facts at all). After # PR #339 the per-scope net total can legitimately be near-zero or # even slightly negative for a small / heavily-SP-discounted scope # in a partial-month window — the old ``total <= 0`` guard would # silently return empty for those, hiding real rows. ``share_pct`` # math below handles ``total_dec <= 0`` separately (returns 0 share) # so per-person / per-service rows still surface their absolute # costs. if fact_count == 0: return { "total_usd": 0.0, "by_person": [], "by_service": [], "by_department": [], "person_count": 0, "fact_count": 0, } # ── By-person: aggregate then enrich with IdentityPerson ───── person_totals: dict[Any, Decimal] = defaultdict(Decimal) for chunk_start, chunk_end in chunks: for r in (await session.execute( select( CmdbCostFact.cost_owner_person_id.label("pid"), func.coalesce(func.sum(amount_col), 0).label("cost"), ) .where(and_(*_chunk_where(chunk_start, chunk_end))) .group_by(CmdbCostFact.cost_owner_person_id), )).all(): person_totals[r.pid] += _dec(r.cost) person_rows = sorted(person_totals.items(), key=_grouped_sort_key) # Enrich with display_name / work_email / department / cost_center. # Only fetch the persons that actually appear in the cost facts — # ``person_ids`` may be much larger (every subordinate), and we # don't want to render a "$0 spend" row for everyone. appearing_person_ids = [pid for pid, _cost in person_rows if pid] person_meta: dict[str, dict[str, str | None]] = {} if appearing_person_ids: meta_rows = (await session.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.cost_center, ).where(IdentityPerson.id.in_(appearing_person_ids)), )).all() for row in meta_rows: person_meta[str(row.id)] = { "display_name": row.display_name, "work_email": row.work_email, "cost_center": row.cost_center, } by_person = [] for pid, cost_dec in person_rows: meta = person_meta.get(str(pid)) if pid else None by_person.append({ "person_id": str(pid) if pid else None, "display_name": (meta or {}).get("display_name"), "work_email": (meta or {}).get("work_email"), "cost_center": (meta or {}).get("cost_center"), "cost_usd": float(round(cost_dec, 2)), "share_pct": ( float(round(100 * cost_dec / total_dec, 2)) if total_dec > 0 else 0.0 ), }) # ── By-service: top 20 (截断在跨月合并之后) ────────────────── service_totals: dict[Any, Decimal] = defaultdict(Decimal) for chunk_start, chunk_end in chunks: for r in (await session.execute( select( CmdbCostFact.service.label("service"), func.coalesce(func.sum(amount_col), 0).label("cost"), ) .where(and_(*_chunk_where(chunk_start, chunk_end))) .group_by(CmdbCostFact.service), )).all(): service_totals[r.service] += _dec(r.cost) by_service = [ { "service": service or "Other", "cost_usd": float(round(cost_dec, 2)), "share_pct": ( float(round(100 * cost_dec / total_dec, 2)) if total_dec > 0 else 0.0 ), } for service, cost_dec in sorted(service_totals.items(), key=_grouped_sort_key)[:20] ] # ── By-department ───────────────────────────────────── dept_totals: dict[Any, Decimal] = defaultdict(Decimal) for chunk_start, chunk_end in chunks: for r in (await session.execute( select( CmdbCostFact.cost_owner_department.label("dept"), func.coalesce(func.sum(amount_col), 0).label("cost"), ) .where(and_(*_chunk_where(chunk_start, chunk_end))) .group_by(CmdbCostFact.cost_owner_department), )).all(): dept_totals[r.dept] += _dec(r.cost) by_department = [ { "department": dept or "", "cost_usd": float(round(cost_dec, 2)), "share_pct": ( float(round(100 * cost_dec / total_dec, 2)) if total_dec > 0 else 0.0 ), "is_unallocated": dept is None or dept == "", } for dept, cost_dec in sorted(dept_totals.items(), key=_grouped_sort_key) ] return { "total_usd": float(round(total_dec, 2)), "by_person": by_person, "by_service": by_service, "by_department": by_department, "person_count": len(by_person), "fact_count": fact_count, } # Tag keys are interpolated into a ``JSON_EXTRACT(..., '$.""')`` # path expression. Even though the surrounding quotes neutralise SQL # injection, an unrestricted key could embed JSON-path metacharacters # (``.``, ``[``, ``]``, ``*``, double-quote, backslash) and either # silently return NULL (no match) or trip TiDB's path parser. AWS / # Aliyun / GCP tag keys are restricted to letters / digits / ``_`` / # ``-`` / ``.`` / ``:`` / ``/``; we accept that union plus ``+`` and # ``@`` (Azure resource tag spec) and reject anything else. Length # capped at 128 to mirror the AWS limit. _TAG_KEY_RE = re.compile(r"^[A-Za-z0-9_\-.:/+@]{1,128}$") _TAG_VALUE_MAX = 256 # Hard SQL ``LIMIT`` cap on the per-resource latest-non-empty-tags # scan in ``top_resources``. Module-level so tests can monkeypatch # it down to exercise the LIMIT-hit graceful-degradation path # without inflating the test DB to 50k rows. Production default is # 50k rows × ~5 KiB worst-case JSON ≈ 250 MiB upper bound, well # under the API worker's budget. Early-termination in Python # typically stops the cursor on the first few hundred rows so the # cap only fires on pathological tag-sparse + high-volume tenants. TOP_RESOURCES_TAG_SCAN_HARD_CAP = 50_000 def _validate_tag_filter( tag_key: str | None, tag_value: str | None, ) -> tuple[str, str] | None: """Validate and normalise a (tag_key, tag_value) filter pair. Returns ``None`` when no filter is requested. Raises ``ValueError`` when only one of the two is provided, or either fails validation — callers map this to a 400 response. """ if tag_key is None and tag_value is None: return None if not tag_key or not tag_value: raise ValueError( "tag_key and tag_value must be provided together", ) if not _TAG_KEY_RE.match(tag_key): raise ValueError( f"invalid tag_key {tag_key!r}: must match " f"{_TAG_KEY_RE.pattern}", ) if len(tag_value) > _TAG_VALUE_MAX: raise ValueError( f"tag_value exceeds {_TAG_VALUE_MAX} chars", ) return tag_key, tag_value def _tag_filter_predicate(tag_key: str, tag_value: str) -> Any: """Return a SQLAlchemy predicate matching rows where ``tags[tag_key] == tag_value``. Uses MySQL/TiDB's ``JSON_EXTRACT`` + ``JSON_UNQUOTE`` so the comparison is on the bare scalar value (without the wrapping JSON double-quotes). ``tag_key`` MUST have been validated by ``_validate_tag_filter`` first — this helper interpolates it directly into the JSON-path string. **Safety contract (do not loosen ``_TAG_KEY_RE`` without re-auditing this site)**: the path ``$.""`` relies on ``_TAG_KEY_RE`` rejecting JSON-path metacharacters that would break key quoting — specifically ``"`` (would close the member-name quote) and ``\\`` (path escape). Other path metacharacters (``.``, ``[``, ``]``, ``*``, ``$``, space) are inert INSIDE a double-quoted member name per the MySQL JSON-path spec, but they're still rejected by the regex for defence-in-depth and to avoid surprising semantics if a future SQL engine treats them differently. """ # Defence-in-depth: a future caller might forget to invoke # ``_validate_tag_filter`` first. Re-validate here so the # interpolation can never run on an unvetted string regardless # of call-site discipline. Cheap (one regex match) and the # AssertionError fires loudly on the bad path. if not _TAG_KEY_RE.match(tag_key): raise ValueError( f"_tag_filter_predicate refused unvalidated tag_key " f"{tag_key!r} — callers MUST run _validate_tag_filter first", ) path = f'$."{tag_key}"' return ( func.json_unquote(func.json_extract(CmdbCostFact.tags, path)) == tag_value ) async def list_tag_keys( session: AsyncSession, *, start: date, end: date, integration_id: str | None = None, department: str | None = None, limit: int = 200, ) -> list[dict[str, Any]]: """Discover the set of user-defined tag keys observed in the cost window, with per-key fact counts. Powers the "filter by tag" autocomplete on the dashboard. Pulls the ``tags`` JSON column row-by-row and aggregates keys in Python — ``JSON_KEYS`` pushdown to TiKV isn't implemented so server-side aggregation is the same cost while adding cross- engine complexity. **Memory bounding** is critical: the prod ``cmdb_cost_facts`` table holds ~10M+ rows and a no-filter call could otherwise OOM the API process. Two-layer cap: 1. Hard SQL ``LIMIT`` (``_TAG_KEYS_SCAN_HARD_CAP``) caps the row stream at ~50 MiB worst-case payload regardless of caller window / tenant filter. 2. ``ORDER BY date DESC`` biases the sample toward recent tag-set composition so a key that was deprecated 6 months ago doesn't dominate the autocomplete when newer keys are in active use. Combined with the ``can_view_finance`` RBAC gate on the calling router endpoint, the surface is not abusable. ``fact_count`` reflects the SAMPLE, not the full window total — the absolute number isn't load-bearing (it's only used to rank the autocomplete). Documented in the field description on the HTTP / MCP surface. """ # ``tags IS NOT NULL`` alone is NOT enough — ``ingest_cost_facts`` # writes ``fact.tags or {}`` so untagged rows persist as JSON # ``{}``, NOT NULL. Without the JSON_LENGTH filter we'd scan # millions of empty-dict rows and discard them in Python. where_clauses = [ CmdbCostFact.date >= start, CmdbCostFact.date <= end, CmdbCostFact.tags.is_not(None), func.json_length(CmdbCostFact.tags) > 0, ] if integration_id: where_clauses.append(CmdbCostFact.integration_id == integration_id) dept_clause = await _department_filter_clause(session, department) if dept_clause is not None: where_clauses.append(dept_clause) # No-tenant variant is intentionally lower-capped (5k vs 50k): # cross-tenant cost-fact scans pull JSON payloads from every # account at once, which can OOM the API worker on a 10M+ row # prod table even with the date window. The single-tenant # variant gets the larger cap since it's bounded by a single # account's daily volume. _TAG_KEYS_SCAN_HARD_CAP = 50_000 if integration_id else 5_000 res = await session.execute( select(CmdbCostFact.tags) .where(and_(*where_clauses)) .order_by(CmdbCostFact.date.desc()) .limit(_TAG_KEYS_SCAN_HARD_CAP), ) key_counts: dict[str, int] = defaultdict(int) for row in res: td = row.tags or {} if not isinstance(td, dict): continue for k in td: if not isinstance(k, str) or not k: continue key_counts[k] += 1 ranked = sorted( key_counts.items(), key=lambda kv: (-kv[1], kv[0]), )[:limit] return [{"key": k, "fact_count": c} for k, c in ranked] async def top_resources( session: AsyncSession, *, start: date, end: date, integration_id: str | None = None, department: str | None = None, resource_group: str | None = None, search: str | None = None, tag_key: str | None = None, tag_value: str | None = None, limit: int = 100, measure: str = "list", session_factory: Any | None = None, ) -> list[dict[str, Any]]: """Top resources by cost over a window, with sparkline-ready 14-day daily series. Used by the dashboard's "Resources" tab — every row carries enough context to open the detail drawer without a follow-up fetch. ``tag_key`` / ``tag_value`` (both-or-neither) narrows the result to rows whose ``CmdbCostFact.tags[tag_key] == tag_value`` — used by the "filter by tag" UX on the dashboard. """ # 口径路由(权限判定在 router 层完成;这里只选列)。研发口径 = # COALESCE(list, invoice) 过渡语义,见 cost_view 模块 docstring。 amount_col = fact_amount_expr(measure) tag_filter = _validate_tag_filter(tag_key, tag_value) # Non-date predicates live in their own list so the month-chunked # pass can re-use them verbatim with its OWN date bounds. Do NOT go # back to filtering the combined list by identity # (``c is not where_clauses[0]``): that silently depends on the date # bounds staying at indices 0 and 1, and the predicates it would drop # on a reorder are the SCOPING ones — integration and department. filter_clauses: list[Any] = [CmdbCostFact.resource_id.is_not(None)] if integration_id: filter_clauses.append(CmdbCostFact.integration_id == integration_id) dept_clause = await _department_filter_clause(session, department) if dept_clause is not None: filter_clauses.append(dept_clause) if resource_group: filter_clauses.append(CmdbCostFact.resource_group == resource_group) if tag_filter is not None: filter_clauses.append(_tag_filter_predicate(*tag_filter)) if search: like = f"%{search}%" filter_clauses.append( or_( CmdbCostFact.resource_id.ilike(like), CmdbCostFact.resource_name.ilike(like), CmdbCostFact.resource_type.ilike(like), ), ) # NOTE: there is deliberately no combined date+filter list any more. # The only aggregate here is the month-chunked one below, and it # supplies its own per-chunk bounds; the sparkline pass further down # builds ``spark_where`` for its own narrower window. # First pass: top resources by total cost — MONTH-CHUNKED. # # TiDB 8175 防线(项目规则,非可选). This was the one wide GROUP BY # in this file without it (#2133): ``/cmdb/cost/resources`` accepts a # 365-day window, groups by the HIGHEST-cardinality dimension there # is (resource_id), and migration 431's covering index does not # include that key — so the aggregate falls back to wide rows over # ~10M rows in one statement. The 2026-07-28 production incident # blew the single-SQL memory quota with 6 months / ~5.5M rows, and # this endpoint has none of the backstops the others got either: no # ``_cached_read``, no fan-out semaphore. # # Each chunk returns its own top slice rather than every group — # resource_id cardinality is orders of magnitude above the # department / service keys ``breakdown`` merges in full, and # materialising a year of them in Python would just move the memory # problem. The slice is deliberately far wider than the caller's # ``limit``; a chunk that actually hits it is logged, because a # silent cap reads as "these are the top resources" when it is not. per_chunk_limit = max(limit * _RESOURCE_CHUNK_OVERSAMPLE, 2000) async def _chunk( chunk_session: AsyncSession, c_start: date, c_end: date, ) -> list[Any]: res = await chunk_session.execute( select( CmdbCostFact.resource_id, CmdbCostFact.integration_id, func.max(CmdbCostFact.resource_name).label("name"), func.max(CmdbCostFact.resource_type).label("type"), func.max(CmdbCostFact.resource_group).label("rg"), func.max(CmdbCostFact.region).label("region"), func.max(CmdbCostFact.service).label("service"), func.max( CmdbCostFact.cost_owner_person_id, ).label("owner_person_id"), func.max(CmdbCostFact.cost_center).label("cost_center"), func.coalesce(func.sum(amount_col), 0).label("total"), ) .where(and_( CmdbCostFact.date >= c_start, CmdbCostFact.date <= c_end, *filter_clauses, )) .group_by( CmdbCostFact.resource_id, CmdbCostFact.integration_id, ) .order_by( func.sum(amount_col).desc(), CmdbCostFact.integration_id.asc(), CmdbCostFact.resource_id.asc(), ) .limit(per_chunk_limit), ) out = list(res.all()) if len(out) >= per_chunk_limit: logger.warning( "top_resources: month chunk %s..%s hit the %d-group cap; " "a resource ranked below it every month could be missing " "from the merged top-%d", c_start, c_end, per_chunk_limit, limit, ) return out merged: dict[tuple[Any, Any], dict[str, Any]] = {} for chunk_rows in await _run_month_chunks( session, _month_chunks(start, end), _chunk, session_factory, ): for r in chunk_rows: key = (r.resource_id, r.integration_id) acc = merged.get(key) if acc is None: merged[key] = { "resource_id": r.resource_id, "integration_id": r.integration_id, "name": r.name, "type": r.type, "rg": r.rg, "region": r.region, "service": r.service, "owner_person_id": r.owner_person_id, "cost_center": r.cost_center, "total": _dec(r.total), } continue # Money accumulates in Decimal (project rule); the # descriptive columns are MAX() per chunk, so keep the first # non-null seen — identical to what a single MAX() over the # whole window would return for a stable resource. acc["total"] += _dec(r.total) for col in ( "name", "type", "rg", "region", "service", "owner_person_id", "cost_center", ): if acc[col] is None: acc[col] = getattr(r, col if col != "type" else "type") # Same deterministic ordering the single-statement version had: # cost desc, then (integration_id, resource_id) so two resources on # an identical flat rate don't swap places between refreshes. rows = [ SimpleNamespace(**v) for v in sorted( merged.values(), key=lambda v: ( -v["total"], str(v["integration_id"]), str(v["resource_id"]), ), )[:limit] ] if not rows: return [] # Second pass: 14-day spark for each top resource. Bound the # window to the trailing 14 days of the request range so a # 90-day backfill view doesn't draw a 90-bar sparkline. # # KEY by ``(integration_id, resource_id)`` — NOT ``resource_id`` # alone. Cloud resource IDs aren't guaranteed unique across # integrations (e.g. two Aliyun tenants can both have an instance # ``i-bp1...``; Azure resource paths embed subscription id but # collisions across mismatched-tenant migrations have been # observed); keying by resource_id only would silently sum two # different resources' daily costs into one sparkline. # Apply the same scoping filters used by the first pass so the # spark sees the same data as the totals row. spark_end = end spark_start = max(start, end - timedelta(days=13)) rid_pairs = {(str(r.integration_id), r.resource_id) for r in rows} spark_where = [ CmdbCostFact.date >= spark_start, CmdbCostFact.date <= spark_end, CmdbCostFact.resource_id.in_({rid for _, rid in rid_pairs}), ] if integration_id: spark_where.append(CmdbCostFact.integration_id == integration_id) if resource_group: spark_where.append(CmdbCostFact.resource_group == resource_group) spark_res = await session.execute( select( CmdbCostFact.integration_id, CmdbCostFact.resource_id, CmdbCostFact.date, func.coalesce( func.sum(amount_col), 0, ).label("daily"), ) .where(and_(*spark_where)) .group_by( CmdbCostFact.integration_id, CmdbCostFact.resource_id, CmdbCostFact.date, ), ) sparks: dict[tuple[str, str], dict[date, float]] = defaultdict(dict) for r in spark_res: sparks[(str(r.integration_id), r.resource_id)][r.date] = ( float(r.daily or 0) ) # Zero-fill the spark grid so frontend doesn't need null handling. spark_dates = [ spark_start + timedelta(days=i) for i in range((spark_end - spark_start).days + 1) ] # Third pass: per-(integration_id, resource_id) latest non-empty # tags dict. ``CmdbCostFact.tags`` is a JSON column so MAX() / # GROUP_CONCAT don't work on it cleanly across MySQL + TiDB. # Strategy: # 1. Hard SQL ``LIMIT`` bounds worst-case memory regardless of # window size or tag heaviness. Worst case payload at the # cap = 50k × ~5 KiB JSON ≈ 250 MiB; in practice well under # because we stream the cursor and break early. # 2. ``ORDER BY date DESC`` biases toward recent rows so the # first hit per resource is the most recent tag dict. # 3. ``tags IS NOT NULL`` filter pushes the "no tags ever" # rows out at the DB layer instead of streaming JSON # payloads we'd immediately discard. # 4. Python early-terminates as soon as every topN resource # has a tag dict captured. On tag-rich tenants this typically # lands on the first few hundred rows. # Read the module-level constant at call time (NOT import time) # so a test that monkeypatches ``TOP_RESOURCES_TAG_SCAN_HARD_CAP`` # actually affects the SQL ``.limit(...)`` clause. _TAG_SCAN_HARD_CAP = TOP_RESOURCES_TAG_SCAN_HARD_CAP target_keys = {(str(i), r) for (i, r) in rid_pairs} tags_by_key: dict[tuple[str, str], dict[str, Any]] = {} # ``tags IS NOT NULL`` + ``JSON_LENGTH(tags) > 0`` — the former # alone matches every row (ingest writes {} for untagged # facts), wasting the SQL LIMIT on empty payloads. tags_where = [ CmdbCostFact.date >= start, CmdbCostFact.date <= end, CmdbCostFact.resource_id.in_({rid for _, rid in rid_pairs}), CmdbCostFact.tags.is_not(None), func.json_length(CmdbCostFact.tags) > 0, ] if integration_id: tags_where.append(CmdbCostFact.integration_id == integration_id) if resource_group: tags_where.append(CmdbCostFact.resource_group == resource_group) tags_res = await session.execute( select( CmdbCostFact.integration_id, CmdbCostFact.resource_id, CmdbCostFact.tags, ) .where(and_(*tags_where)) .order_by(CmdbCostFact.date.desc()) .limit(_TAG_SCAN_HARD_CAP), ) for tr in tags_res: key = (str(tr.integration_id), tr.resource_id) if key not in target_keys: # Cross-integration resource_id collision — the # ``resource_id.in_(...)`` predicate doesn't constrain # integration, so two tenants with the same resource_id # can both appear in the stream. Drop the ones that # weren't in our topN set. continue if key in tags_by_key: # Already captured the latest non-empty entry for this # resource — older rows can't override. continue td = tr.tags or {} if not td: continue if not isinstance(td, dict): # Defensive — the column is JSON-typed but a buggy ingest # could write a non-object payload. Skip rather than # raise; the row still gets a ``tags={}`` downstream. continue tags_by_key[key] = td if len(tags_by_key) >= len(target_keys): # Every topN resource has a tag dict — early termination # short-circuits the rest of the cursor read. break out: list[dict[str, Any]] = [] for r in rows: spark_key = (str(r.integration_id), r.resource_id) out.append({ "resource_id": r.resource_id, "integration_id": str(r.integration_id), "resource_name": r.name, "resource_type": r.type, "resource_group": r.rg, "region": r.region, "service": r.service, "owner_person_id": ( str(r.owner_person_id) if r.owner_person_id else None ), "cost_center": r.cost_center, "total_usd": round(float(r.total or 0), 2), "spark": [ round(sparks[spark_key].get(d, 0.0), 2) for d in spark_dates ], "tags": tags_by_key.get(spark_key, {}), }) return out async def timeseries( session: AsyncSession, *, start: date, end: date, group_by: str, integration_id: str | None = None, scope_id: str | None = None, department: str | None = None, top: int = 10, measure: str = "list", session_factory: Any | None = None, ) -> dict[str, Any]: """Daily series broken out by ``group_by``, with top-N + "other". Returns ``{dates: [...], series: [{key, label, points: [...]}], total_per_day: [...]}``. Zero-filled — the frontend can plot directly without nullish checks. ``session_factory`` (optional): same contract as :func:`breakdown` — bounded-concurrency month chunks on their own sessions when passed, sequential on ``session`` when not. See :func:`_run_month_chunks`. """ # 口径路由(权限判定在 router 层完成;这里只选列)。研发口径 = # COALESCE(list, invoice) 过渡语义,见 cost_view 模块 docstring。 amount_col = fact_amount_expr(measure) # ``provider`` groups on the joined ``integrations.provider`` (see the # matching note in ``breakdown``). The two grouped queries below get # the ``Integration`` join; the SP sub-query filters only cost_facts. is_provider_dim = group_by == "provider" if is_provider_dim: col = Integration.provider else: col = GROUP_BY_COLUMNS.get(group_by) if col is None: raise ValueError( f"unknown group_by={group_by!r}; supported: " f"{sorted([*GROUP_BY_COLUMNS, 'provider'])}", ) base_clauses = [] if integration_id: base_clauses.append(CmdbCostFact.integration_id == integration_id) base_clauses.extend(await _scope_filter_clauses(session, scope_id)) dept_clause = await _department_filter_clause(session, department) if dept_clause is not None: base_clauses.append(dept_clause) def _maybe_join_provider(stmt): if is_provider_dim: return stmt.select_from(CmdbCostFact).join( Integration, Integration.id == CmdbCostFact.integration_id, ) return stmt # Mirror ``breakdown()``'s Savings-Plans handling so the two surfaces # on the Cost-Analysis screen agree. SP discount/fee rows # (``auto_owner_source='savings_plan'``, NULL cost-owner) are org-level # netting, not user-attributable spend; on the allocation dims # (owner/department/cost_center) they otherwise fold into the NULL # "未分摊" series, so the chart's unallocated band would carry the # (usually large negative) SP net while the breakdown KPI tile — which # already excludes SP — would not. EXCLUDE SP from the grouped # top-N/daily queries and re-add it as a single pinned series below. not_sp = func.coalesce(CmdbCostFact.auto_owner_source, "") != "savings_plan" # TiDB 8175 防线:三个聚合按月分块(``_month_chunks``,项目规则, # 非可选)——单条 SQL 的日期窗口 ≤1 个自然月。top-N 键在全窗口 # 合计合并**之后**才截断,所以跨月分组键的排名与未分块一致; # 各月日粒度行合并时日期天然不相交。分块间的执行策略见 # :func:`_run_month_chunks`;金额一律 Decimal 累加。 async def _chunk( chunk_session: AsyncSession, chunk_start: date, chunk_end: date, ) -> tuple[ list[tuple[Any, Decimal]], list[tuple[Any, date, Decimal]], list[tuple[date, Decimal]], ]: chunk_clauses = [ CmdbCostFact.date >= chunk_start, CmdbCostFact.date <= chunk_end, *base_clauses, ] # Window totals per key — merged across chunks BEFORE the # top-N cut so a key's rank reflects its full-window total. totals_q = await chunk_session.execute( _maybe_join_provider(select( col.label("key"), func.coalesce( func.sum(amount_col), 0, ).label("total"), )) .where(and_(*chunk_clauses, not_sp)) .group_by(col), ) totals = [(r.key, _dec(r.total)) for r in totals_q] # Daily breakdown rows; bucketed into top-N/"other" after the # merged ranking is known. daily_q = await chunk_session.execute( _maybe_join_provider(select( col.label("key"), CmdbCostFact.date, func.coalesce( func.sum(amount_col), 0, ).label("daily"), )) .where(and_(*chunk_clauses, not_sp)) .group_by(col, CmdbCostFact.date), ) daily = [(r.key, r.date, _dec(r.daily)) for r in daily_q] # Dedicated SP daily sum, independent of ``group_by`` (matches # breakdown()'s SP stat). Only rendered when the window actually # has SP rows, so GCP-only tenants don't get a spurious all-zero # band. sp_daily_q = await chunk_session.execute( select( CmdbCostFact.date, func.coalesce( func.sum(amount_col), 0, ).label("daily"), ) .where( and_( *chunk_clauses, CmdbCostFact.auto_owner_source == "savings_plan", ), ) .group_by(CmdbCostFact.date), ) return totals, daily, [(r.date, _dec(r.daily)) for r in sp_daily_q] key_totals: dict[Any, Decimal] = defaultdict(Decimal) daily_map: dict[tuple[Any, date], Decimal] = defaultdict(Decimal) sp_by_date: dict[date, Decimal] = defaultdict(Decimal) for totals, daily, sp_daily in await _run_month_chunks( session, _month_chunks(start, end), _chunk, session_factory, ): for key, cost in totals: key_totals[key] += cost for key, day, cost in daily: daily_map[(key, day)] += cost for day, cost in sp_daily: sp_by_date[day] += cost # 与 breakdown() 同一条裁剪:研发口径下恒 0 的序列只可能由全部被 # 白名单排除的行构成,留下来就是给堆叠图挂一条压在 0 上的死图例。 # 实付口径不裁(净额为 0 的服务是真实存在的)。 if measure != COST_MEASURE_INVOICE: dropped = {k for k, v in key_totals.items() if v == 0} key_totals = {k: v for k, v in key_totals.items() if k not in dropped} # ``daily_map`` 必须一起裁 —— 否则被裁掉的 key 会掉进下面的 # ``other_map``:① ``other_map`` 是 defaultdict,``+= 0`` 就把日期 # 键建出来,``if other_map:`` 于是为真,凭空多出一条**全零的 # 「其他」**图例 —— 正是这段裁剪想消灭的死图例,只是换了个名字; # ② 窗口合计恰为 0、但每日有正负相抵的 key(阿里云把升降配退款 # 归在 usage 桶,签名用量行确实存在),它真实的日尖峰会被记到 # 「其他」头上,读者在堆叠图上看到一个来路不明的 +X/−X。 # breakdown() 没有这个问题:它在切 top-N/other **之前**就裁了。 daily_map = { (k, d): v for (k, d), v in daily_map.items() if k not in dropped } top_keys = [ key for key, _cost in sorted(key_totals.items(), key=_grouped_sort_key)[:top] ] series_map: dict[Any, dict[date, Decimal]] = defaultdict(dict) other_map: dict[date, Decimal] = defaultdict(Decimal) for (key, d), amount in daily_map.items(): if key in top_keys: series_map[key][d] = amount else: other_map[d] += amount days = [ start + timedelta(days=i) for i in range((end - start).days + 1) ] # Round-27 pre-launch (2026-05-25): backend returns raw key + flags # (``is_unallocated``, ``is_other``) and an empty ``label`` so the # frontend i18ns the user-facing string. Mirrors ``breakdown()``'s # contract above. Previously ``label`` carried hardcoded zh-CN # ``"(空)"`` / ``"其他"`` that en-US clients (web + MCP) could not # translate. series_out = [] for key in top_keys: series_out.append({ "key": key if key is not None else "", "label": key if key else "", "is_unallocated": key is None or key == "", "is_other": False, "is_discount": False, "points": [ float(round(series_map[key].get(d, Decimal(0)), 2)) for d in days ], }) if other_map: series_out.append({ "key": "__other__", "label": "", "is_unallocated": False, "is_other": True, "is_discount": False, "points": [ float(round(other_map.get(d, Decimal(0)), 2)) for d in days ], }) # ``owner`` keys are raw person UUIDs — resolve to display names so # the stacked timeseries legend reads as people (mirrors breakdown()). # Runs BEFORE the SP series is appended so the literal "Savings Plans" # key is never coerced to a UUID by ``_resolve_owner_labels``. if group_by == "owner": owner_labels = await _resolve_owner_labels( session, [str(s["key"]) for s in series_out if s["key"] and not s["is_other"] and not s["is_unallocated"]], ) for s in series_out: if s["key"] and not s["is_other"] and not s["is_unallocated"]: s["label"] = owner_labels.get(str(s["key"]), str(s["key"])) # Pin Savings Plans as its own band (mirrors breakdown()'s pinned SP # group). ``is_unallocated=False`` — it is netting, not missing-owner — # and ``is_discount`` drives the FE's green affordance when negative. sp_total = sum(sp_by_date.values(), Decimal(0)) if sp_by_date else Decimal(0) # ``sp_total != 0`` mirrors breakdown()'s pinned-group guard: under the # engineering caliber every SP row aggregates to 0 (commitment buckets # are out of the list whitelist), and an all-zero band would add a dead # legend entry to every stacked chart. See cost_view 模块 docstring. if sp_by_date and sp_total != 0: # ``is_discount`` semantics MUST match breakdown()'s pinned SP group # (``sp_sum < 0``) — it drives the FE's green affordance and is only # meaningful when the SP net is negative (the common discount case); # a rare positive-net SP window renders neutral on BOTH surfaces. series_out.append({ "key": "Savings Plans", "label": "Savings Plans", "is_unallocated": False, "is_other": False, "is_discount": sp_total < 0, "points": [ float(round(sp_by_date.get(d, Decimal(0)), 2)) for d in days ], }) total_per_day = [ round( sum(s["points"][i] for s in series_out), 2, ) for i in range(len(days)) ] return { "dates": [d.isoformat() for d in days], "series": series_out, "total_per_day": total_per_day, } async def forecast_month_end( session: AsyncSession, *, integration_id: str | None = None, department: str | None = None, today: date | None = None, measure: str = "list", ) -> dict[str, Any]: """Project month-end spend using a simple two-method blend. * **Linear**: MTD spend × (days_in_month / days_elapsed). Sensitive to early-month spikes and reserved-instance front-loading. * **Trailing**: 14d trailing daily mean × days_remaining + MTD. Smoothes intra-month volatility. The blend is the average of the two; we surface both individually so the dashboard can show "the bound." MTD is also returned for context. ``today`` defaults to ``datetime.now(UTC).date()`` to match the UTC-anchored ``CmdbCostFact.date`` column. Process-local ``date.today()`` would be off by a day on UTC-deployed workers serving SH-anchored cron data. """ # 口径路由(权限判定在 router 层完成;这里只选列)。研发口径 = # COALESCE(list, invoice) 过渡语义,见 cost_view 模块 docstring。 amount_col = fact_amount_expr(measure) from datetime import UTC today = today or datetime.now(UTC).date() # Anchor MTD on YESTERDAY, not today. Cost facts are ingested at # 04:00 UTC on the prior day's data; today's row is empty until # tomorrow. Using ``today`` as the upper bound underestimates the # projection by ~1/days_elapsed (worst near the start of the # month). Matches the anomaly-detection path which also uses # ``today - 1`` for the same reason. last_complete_day = today - timedelta(days=1) month_start = today.replace(day=1) if today.month == 12: next_month_start = today.replace( year=today.year + 1, month=1, day=1, ) else: next_month_start = today.replace(month=today.month + 1, day=1) days_in_month = (next_month_start - month_start).days # Round-22 audit follow-up: on day-1 of a month, ``last_complete_ # day`` is the last day of the PRIOR month and the cron has not yet # ingested any fact for the current month (today's data lands at # 04:00 UTC tomorrow). The earlier fallback "if last_complete_day # < month_start: mtd_upper = today" then produced a forecast # blending the prior month's trailing mean × 30 against an # MTD-of-zero linear projection — a spurious "30× yesterday's # spend" projection chart on every December 1st leadership view. # Honest behaviour: when the current month has no complete day of # data yet, return zero MTD + a null forecast with a reason flag # so the FE can render "month just started — no projection until # the first complete day". This matches how `detect_anomalies` # already short-circuits on insufficient data. if last_complete_day < month_start: return { "month_start": month_start.isoformat(), "today": today.isoformat(), "days_elapsed": 0, "days_remaining": (next_month_start - today).days, "days_in_month": days_in_month, "mtd_usd": 0.0, "trail14_daily_mean_usd": 0.0, "forecast_linear_usd": None, "forecast_trailing_usd": None, "forecast_blend_usd": None, "reason": "month_just_started", } mtd_upper = last_complete_day days_elapsed = (mtd_upper - month_start).days + 1 # ``days_remaining`` must count from TODAY (the partially-elapsed # day), not from ``mtd_upper`` (last complete day). Otherwise we # over-project the trailing forecast by one day's smoother mean # — small but compounding across the month. When ``mtd_upper`` is # before today (typical), there's exactly one "in flight" day # between them. days_remaining = max(0, (next_month_start - today).days) dept_clause = await _department_filter_clause(session, department) where_clauses = [ CmdbCostFact.date >= month_start, CmdbCostFact.date <= mtd_upper, ] if integration_id: where_clauses.append(CmdbCostFact.integration_id == integration_id) if dept_clause is not None: where_clauses.append(dept_clause) mtd_q = await session.execute( select(func.coalesce(func.sum(amount_col), 0)) .where(and_(*where_clauses)), ) mtd = float(mtd_q.scalar() or 0) # Trailing 14-day window for the smoother — also anchored on the # last complete day to match MTD's data availability. trail_start = mtd_upper - timedelta(days=13) trail_clauses = [ CmdbCostFact.date >= trail_start, CmdbCostFact.date <= mtd_upper, ] if integration_id: trail_clauses.append(CmdbCostFact.integration_id == integration_id) if dept_clause is not None: trail_clauses.append(dept_clause) trail_q = await session.execute( select(func.coalesce(func.sum(amount_col), 0)) .where(and_(*trail_clauses)), ) trail_sum = float(trail_q.scalar() or 0) trail_days = (mtd_upper - trail_start).days + 1 trail_mean = trail_sum / trail_days if trail_days > 0 else 0.0 linear = ( mtd / days_elapsed * days_in_month if days_elapsed > 0 else 0.0 ) trailing = mtd + trail_mean * days_remaining blend = (linear + trailing) / 2.0 return { "month_start": month_start.isoformat(), "today": today.isoformat(), "days_elapsed": days_elapsed, "days_remaining": days_remaining, "days_in_month": days_in_month, "mtd_usd": round(mtd, 2), "trail14_daily_mean_usd": round(trail_mean, 2), "forecast_linear_usd": round(linear, 2), "forecast_trailing_usd": round(trailing, 2), "forecast_blend_usd": round(blend, 2), } async def detect_anomalies( session: AsyncSession, *, days: int = 30, integration_id: str | None = None, integration_ids: Sequence[str] | None = None, department: str | None = None, min_impact_usd: float = 50.0, z_threshold: float = 2.5, limit: int | None = 1000, measure: str = "list", as_of_date: date | None = None, resource_keys: set[tuple[str, str]] | None = None, chunk_days: int | None = None, ) -> list[dict[str, Any]]: """Surface days where a resource's cost spiked beyond its norm. Per-resource median + IQR over the prior ``days`` days; any day above ``median + z_threshold × IQR`` AND with absolute USD impact above ``min_impact_usd`` is reported. Output sorted by impact USD descending so the dashboard's first row is the biggest fire. Pure stdlib statistics — no numpy / scipy. Volumes (one-org cost facts) trivially fit in process memory. Window anchor: ``yesterday`` (UTC). The cron writes ``CmdbCostFact`` rows for *yesterday* at 04:00 UTC, so today's row is reliably empty until the next cron tick. Anchoring on today would feed a 0-cost row into the per-resource median / IQR distribution, artificially raising the spike threshold and masking real spikes for resources with strict daily activity. Anchoring on yesterday makes the analysis read against the data that's actually been ingested. 口径一致性（#2079 同款闸，2026-07-28）：尖峰是「比值」判定，分子 分母必须同口径。研发口径（list）若按行 COALESCE(list, invoice)， 回填边界横跨窗口时会产生混合基线（基线净额 vs 当日 list —— 1.65× 假尖峰的生产机制，见 lifecycle_service 的 cron 版闸）。这里对每个 资源窗口做「全有才用 list、否则整组净额」的 all-rows-or-nothing 路由：窗口内**每一行**都携带 list 才用 list 序列，任一行缺 list 则整组回落净额 —— 两侧永远同列，比值不受回填进度影响。 ``measure='invoice'`` 恒读净额列，不受影响。 """ from datetime import UTC from sqlalchemy import case # See docstring — yesterday, not today. today = (as_of_date or datetime.now(UTC).date()) - timedelta(days=1) start = today - timedelta(days=days - 1) where_clauses = [ CmdbCostFact.resource_id.is_not(None), ] if integration_id and integration_ids: raise ValueError("integration_id and integration_ids are mutually exclusive") if integration_id: where_clauses.append(CmdbCostFact.integration_id == integration_id) elif integration_ids: where_clauses.append(CmdbCostFact.integration_id.in_(integration_ids)) if measure != COST_MEASURE_INVOICE: where_clauses.append(engineering_row_clause()) dept_clause = await _department_filter_clause(session, department) if dept_clause is not None: where_clauses.append(dept_clause) # Grain MUST be (integration_id, resource_id, date) — NOT # (resource_id, date). Cloud resource IDs are not unique across # integrations (two Aliyun tenants can both have ``i-bp1...``; Azure # paths collide across mismatched-tenant migrations — the sibling # ``top_resources`` documents these as observed in prod and keys on # ``(integration_id, resource_id)`` for exactly this reason). Keying # the anomaly series on ``resource_id`` alone summed two different # resources' daily costs into one synthetic distribution, corrupting # the median/IQR and either fabricating a spike or masking one. Pull # ``integration_id`` into the GROUP BY + bucket key so each physical # resource gets its own series. # 双列取数:净额(= invoice 列)与 list 列分开聚合,外加「缺 list # 行数」。口径选择推迟到 Python 侧的 per-resource 整组闸 —— 不再 # 用逐行 COALESCE 表达式(那正是混合基线的来源)。 # # TiDB 8175 防线(``_month_chunks``,项目规则,非可选):这是本文件 # 里最宽的 GROUP BY(integration × resource × date),窗口可达 120 # 天。按月分块把单条 SQL 的日期窗口压到 ≤1 个自然月。合并是纯拼接 # ——(integration, resource, date) 的行只可能落在一个分块里(分块 # 日期不相交),所以整组口径闸(全窗口 all-rows-or-nothing)与 # ``missing_list`` 计数仍然是全窗口语义:下面的闸跑在拼接后的 # ``raw_points`` 上,不在分块内。 by_resource: dict[ tuple[str, str], list[tuple[date, float, float | None, int, dict[str, Any]]], ] = defaultdict(list) # The default calendar-month chunks are right for normal dashboard # queries. Consumers that sweep every active cloud account at once may # request a smaller cap: TiDB's GROUP BY memory scales with both dates # and accounts, so one calendar month can still exceed its limit. if chunk_days is None: chunks = _month_chunks(start, today) else: chunk_days = max(1, int(chunk_days)) chunks = [] chunk_start = start while chunk_start <= today: chunk_end = min(today, chunk_start + timedelta(days=chunk_days - 1)) chunks.append((chunk_start, chunk_end)) chunk_start = chunk_end + timedelta(days=1) for chunk_start, chunk_end in chunks: res = await session.execute( select( CmdbCostFact.integration_id, CmdbCostFact.resource_id, CmdbCostFact.date, func.coalesce( func.sum(CmdbCostFact.amount_usd), 0, ).label("daily_net"), func.sum(CmdbCostFact.amount_usd_list).label("daily_list"), func.sum( case( (CmdbCostFact.amount_usd_list.is_(None), 1), else_=0, ), ).label("missing_list"), func.max(CmdbCostFact.resource_name).label("name"), func.max(CmdbCostFact.service).label("service"), func.max(CmdbCostFact.resource_group).label("rg"), ) .where(and_( CmdbCostFact.date >= chunk_start, CmdbCostFact.date <= chunk_end, *where_clauses, )) .group_by( CmdbCostFact.integration_id, CmdbCostFact.resource_id, CmdbCostFact.date, ), ) for r in res: resource_key = (str(r.integration_id), str(r.resource_id)) # Intelligence first establishes which resources have enough # complete-day coverage. Filter before building the in-process # series so sparse/out-of-scope resources do not consume memory. if resource_keys is not None and resource_key not in resource_keys: continue by_resource[resource_key].append(( r.date, float(r.daily_net or 0), float(r.daily_list) if r.daily_list is not None else None, int(r.missing_list or 0), { "name": r.name, "service": r.service, "integration_id": str(r.integration_id), "rg": r.rg, }, )) out: list[dict[str, Any]] = [] for (_integ_id, rid), raw_points in by_resource.items(): if len(raw_points) < 7: continue # not enough history to call a spike # 整组口径闸(all-rows-or-nothing):仅当该资源窗口内每一天的 # 每一行都携带 list 才用 list 序列;否则整组净额。invoice 请求 # 恒净额。 use_list = measure != COST_MEASURE_INVOICE and all( p[3] == 0 and p[2] is not None for p in raw_points ) points: list[tuple[date, float, dict[str, Any]]] = [ (p[0], (p[2] if use_list else p[1]), p[4]) # type: ignore[misc] for p in raw_points ] # Sort by date so the leave-one-out distribution comparison # below has stable point indexing across runs. points.sort(key=lambda p: p[0]) amounts = [p[1] for p in points] median = statistics.median(amounts) try: q1, _, q3 = statistics.quantiles(amounts, n=4) except statistics.StatisticsError: continue # IQR floor: keep the real IQR when it's meaningful, but the # collapsed-distribution case (resource turned off all window # → IQR = 0) used to floor at ``1e-6`` and let any non-zero # day cross ``threshold = median + z_threshold * 1e-6``, # producing a false-positive that only ``min_impact_usd`` # caught downstream. Take the max of three floors: # 1. real IQR ``q3 - q1`` (the statistically correct value # when the distribution has actual spread); # 2. percentage floor ``abs(median) * 5%`` (handles low-but- # nonzero baselines — a tiny but plausible variation); # 3. absolute floor ``min_impact_usd / z_threshold`` (a # backstop tied to the same impact gate downstream uses, # so the z-score gate never demands a smaller spike than # the impact filter would let through anyway). # Round-2 review caught the previous conditional formula — it # only engaged the absolute floor when ``median == 0`` exactly, # leaving tiny-but-nonzero medians (creep-up baselines) using # the percentage floor alone. ``max(...)`` of all three is # both simpler and uniformly defensive. iqr_basic = q3 - q1 pct_floor = abs(median) * 0.05 absolute_floor = min_impact_usd / max(z_threshold, 1e-6) iqr = max(iqr_basic, pct_floor, absolute_floor, 1e-6) threshold = median + z_threshold * iqr # Per-day evaluation uses a leave-one-out distribution: the # candidate spike day MUST NOT contribute to its own baseline # median / IQR. Including it inflates the median and IQR # proportional to the spike size, masking sustained 2-3-day # spikes (each day inflates the threshold partially shielding # the others). ``statistics.quantiles(data, n=4)`` requires # ``len(data) >= 2``; with ``len(points) >= 7`` already gated # above the leave-one-out distribution always has 6+ points, # well above the API minimum, so we compute the per-candidate # baseline unconditionally and let a ``StatisticsError`` (the # all-equal degenerate distribution) fall back to the all- # points threshold computed once outside the loop. for idx, (d, amount, meta) in enumerate(points): others = amounts[:idx] + amounts[idx + 1 :] try: oq1, _, oq3 = statistics.quantiles(others, n=4) except statistics.StatisticsError: eff_median = median eff_threshold = threshold # Round-49: degenerate-distribution branch — surface # the global IQR (which IS what produced the threshold # here) so downstream dashboards / forensic reads see # the value that was actually used to gate this row. eff_iqr = iqr else: eff_median = statistics.median(others) eff_iqr = max( oq3 - oq1, abs(eff_median) * 0.05, absolute_floor, 1e-6, ) eff_threshold = eff_median + z_threshold * eff_iqr if amount <= eff_threshold: continue impact = amount - eff_median if impact < min_impact_usd: continue out.append({ "resource_id": rid, "date": d.isoformat(), "amount_usd": round(amount, 2), "median_usd": round(eff_median, 2), "impact_usd": round(impact, 2), # Round-49: report the per-candidate IQR that actually # produced ``eff_threshold`` for this row, not the # global IQR (which differs whenever the leave-one-out # distribution shifted q1 / q3 — i.e. on any sustained # spike day where excluding the candidate meaningfully # changed the spread). "iqr_usd": round(eff_iqr, 2), "resource_name": meta["name"], "service": meta["service"], "integration_id": meta["integration_id"], "resource_group": meta["rg"], }) # Stable order with deterministic tiebreaker — two cron / MCP # invocations in succession return the same ranking even when # impact_usd ties. Without the tiebreaker, downstream alerting # idempotency (dedup keyed on the leading anomaly) becomes # non-deterministic. out.sort( key=lambda x: ( -x["impact_usd"], x["resource_id"] or "", x["date"], ), ) # Round-49: bound the result set. Unlike ``/resources`` and # ``/breakdown`` which enforce hard limits in the endpoint, the # raw anomaly list was unbounded — a 30-day window on an org with # many resources and a sustained-spike pattern could return # 10 000+ rows, consuming several MB per response and blocking the # event loop on JSON encoding. Apply the cap AFTER sorting so the # most-impactful spikes are always the ones surfaced. ``None`` # preserves the un-capped behaviour for internal callers # (allocation recompute, forensic walks) that genuinely want # everything. if limit is not None and limit > 0 and len(out) > limit: out = out[:limit] return out # ─── Allocation rule CRUD helpers ─────────────────────────────────── async def list_allocation_rules( session: AsyncSession, ) -> list[dict[str, Any]]: res = await session.execute( select(CmdbCostAllocationRule) .order_by( CmdbCostAllocationRule.priority.asc(), CmdbCostAllocationRule.created_at.asc(), ), ) return [_serialize_rule(r) for r in res.scalars().all()] def _serialize_rule(r: CmdbCostAllocationRule) -> dict[str, Any]: return { "id": str(r.id), "name": r.name, "priority": r.priority, "match_integration_ids": r.match_integration_ids or [], "match_provider": r.match_provider, "match_resource_group": r.match_resource_group, "match_service": r.match_service, "match_tag_key": r.match_tag_key, "match_tag_value": r.match_tag_value, "assign_person_id": ( str(r.assign_person_id) if r.assign_person_id else None ), "assign_department": r.assign_department, "assign_cost_center": r.assign_cost_center, "enabled": bool(r.enabled), "description": r.description, "created_at": r.created_at.isoformat() if r.created_at else None, "updated_at": r.updated_at.isoformat() if r.updated_at else None, } def _validate_rule_shape(payload: dict[str, Any]) -> None: """Static-shape validation of a payload (create OR patch). Only catches errors that don't depend on the rule's prior state: integration-ids must be a clean list-of-non-empty-strings. Combined-state invariants (tag-value-without-key, at-least-one- assign-*) are checked separately AFTER the patch is applied — those need the merged rule to evaluate correctly under PATCH. """ raw_ids = payload.get("match_integration_ids") if raw_ids is not None and not isinstance(raw_ids, list): raise ValueError("match_integration_ids must be a list of strings") if isinstance(raw_ids, list): for v in raw_ids: if not isinstance(v, str) or not v.strip(): raise ValueError( "match_integration_ids entries must be non-empty strings", ) async def create_allocation_rule( session: AsyncSession, *, payload: dict[str, Any], actor_id: str | None, ) -> dict[str, Any]: if not ( payload.get("assign_person_id") or payload.get("assign_department") or payload.get("assign_cost_center") ): raise ValueError( "rule must set at least one of assign_person_id / " "assign_department / assign_cost_center", ) _validate_rule_shape(payload) if ( payload.get("match_tag_value") and not payload.get("match_tag_key") ): raise ValueError( "match_tag_value is meaningful only with match_tag_key — " "set the key as well, or clear the value", ) # Validate assign_person_id refers to a real person — same # reasoning as update_allocation_rule (don't stamp phantom # owners onto facts). if payload.get("assign_person_id"): person = await session.get( IdentityPerson, payload["assign_person_id"], ) if person is None: raise ValueError( f"assign_person_id={payload['assign_person_id']!r} " "does not refer to an existing identity_persons row", ) rule = CmdbCostAllocationRule( id=str(uuid.uuid4()), name=str(payload.get("name") or "未命名规则")[:255], priority=int(payload.get("priority") or 100), match_integration_ids=payload.get("match_integration_ids") or None, match_provider=payload.get("match_provider"), match_resource_group=payload.get("match_resource_group"), match_service=payload.get("match_service"), match_tag_key=payload.get("match_tag_key"), match_tag_value=payload.get("match_tag_value"), assign_person_id=payload.get("assign_person_id"), assign_department=payload.get("assign_department"), assign_cost_center=payload.get("assign_cost_center"), enabled=bool(payload.get("enabled", True)), description=payload.get("description"), created_by_id=actor_id, updated_by_id=actor_id, ) session.add(rule) await session.commit() await session.refresh(rule) return _serialize_rule(rule) _RULE_MUTABLE_FIELDS: frozenset[str] = frozenset({ "name", "priority", "match_integration_ids", "match_provider", "match_resource_group", "match_service", "match_tag_key", "match_tag_value", "assign_person_id", "assign_department", "assign_cost_center", "enabled", "description", }) async def update_allocation_rule( session: AsyncSession, *, rule_id: str, payload: dict[str, Any], actor_id: str | None, trace_id: str | None = None, ip_address: str | None = None, user_agent: str | None = None, ) -> dict[str, Any]: """Patch a rule. Concurrency-safe: row-locks the rule before mutation so two simultaneous PATCHes can't independently land contradictory states. ``payload`` is field-whitelisted via ``_RULE_MUTABLE_FIELDS`` — arbitrary keys (created_by_id, id, updated_at) are silently dropped rather than written through to the row. ``assign_person_id`` is verified to refer to a real ``identity_persons`` row before the commit, so a typo'd UUID fails with a 422 instead of stamping a phantom owner onto every matched fact. Round-49 audit hardening: capture the pre-mutation snapshot and write an audit row inside the SAME transaction as the mutation. Pre-fix the router wrote the audit AFTER the service committed, so a post-commit audit failure left the rule changed with NO trail of who / what / when / from-what-prior-state — a silent erasure of the change history for cost attribution. The audit detail now carries ``old_values`` + ``new_values`` so a rule edit is fully reconstructable from the log, not just "some field on this rule was touched at time T". """ from app.audit.service import AuditService # SELECT FOR UPDATE — every concurrent PATCH on the same rule # serialises through this lock, so the post-mutation # invariant check below sees the row state EVERY writer sees. res = await session.execute( select(CmdbCostAllocationRule) .where(CmdbCostAllocationRule.id == rule_id) .with_for_update(), ) rule = res.scalar_one_or_none() if rule is None: raise LookupError(f"allocation rule {rule_id} not found") # Round-49: snapshot prior state under the row lock so the audit # detail and the change diff see the same "before" picture every # concurrent writer sees. previous = _serialize_rule(rule) # Static-shape invariants on the patch itself (list-shape of # match_integration_ids). Combined-state invariants are # re-checked below AFTER applying the patch onto the locked row, # so a partial patch can't violate them silently. _validate_rule_shape(payload) # Pre-validate assign_person_id so we don't lock + commit then # have analytics queries surface a phantom owner. if "assign_person_id" in payload and payload["assign_person_id"]: person = await session.get( IdentityPerson, payload["assign_person_id"], ) if person is None: raise ValueError( f"assign_person_id={payload['assign_person_id']!r} " "does not refer to an existing identity_persons row", ) for field in _RULE_MUTABLE_FIELDS: if field in payload: setattr(rule, field, payload[field]) rule.updated_by_id = actor_id if not ( rule.assign_person_id or rule.assign_department or rule.assign_cost_center ): raise ValueError( "rule must keep at least one assign_* field non-NULL", ) # Combined-state tag invariant: re-check after patch is applied. if rule.match_tag_value and not rule.match_tag_key: raise ValueError( "match_tag_value is meaningful only with match_tag_key — " "set the key as well, or clear the value", ) new_values = _serialize_rule(rule) # Compute per-field deltas restricted to fields the caller # actually touched so the audit detail stays focused on real # changes (a noop PATCH still writes an audit row, but with an # empty ``changed_fields`` — that's intentional: the attempt # itself is recorded). touched = sorted(set(payload.keys()) & _RULE_MUTABLE_FIELDS) old_values: dict[str, Any] = {} new_changed: dict[str, Any] = {} for k in touched: if previous.get(k) != new_values.get(k): old_values[k] = previous.get(k) new_changed[k] = new_values.get(k) audit = AuditService(session) await audit.log( trace_id=trace_id or "", actor_type="user" if actor_id else "system", actor_user_id=uuid.UUID(actor_id) if actor_id else None, action="cost.allocation_rule.updated", status="success", resource_type="cmdb_cost_allocation_rule", resource_id=rule_id, detail={ "changed_fields": sorted(new_changed.keys()), "old_values": old_values, "new_values": new_changed, }, ip_address=ip_address, user_agent=user_agent, ) await session.commit() return new_values async def delete_allocation_rule( session: AsyncSession, *, rule_id: str, actor_id: str | None = None, trace_id: str | None = None, ip_address: str | None = None, user_agent: str | None = None, ) -> dict[str, Any]: """Delete a rule. Raises ``LookupError`` if the row doesn't exist so the HTTP layer surfaces 404 — silently 200-OK on a DELETE that did nothing would let two concurrent admins both 'succeed' deleting the same rule and never know. Concurrency: row-locked via ``SELECT FOR UPDATE`` so two simultaneous DELETEs on the same id serialise — the loser sees the row gone and raises LookupError instead of silently no-op'ing alongside the winner. Round-49 audit hardening: the full pre-delete snapshot is captured under the row lock and persisted into the audit row's ``detail`` BEFORE the actual ``session.delete(rule)`` runs. Pre-fix the router wrote ``detail={}`` AFTER the delete had already committed, so an admin deleting a critical attribution rule erased not only the rule but also the only record of what the rule looked like — the change history could not be reconstructed at all. Now both writes share one transaction; audit + delete commit together or neither does, and the audit detail carries every field of the deleted rule. Returns the serialized pre-delete snapshot so callers can show the user what was removed (and to make undo / restore feasible from a forensic audit row). """ from app.audit.service import AuditService res = await session.execute( select(CmdbCostAllocationRule) .where(CmdbCostAllocationRule.id == rule_id) .with_for_update(), ) rule = res.scalar_one_or_none() if rule is None: raise LookupError(f"allocation rule {rule_id} not found") snapshot = _serialize_rule(rule) audit = AuditService(session) await audit.log( trace_id=trace_id or "", actor_type="user" if actor_id else "system", actor_user_id=uuid.UUID(actor_id) if actor_id else None, action="cost.allocation_rule.deleted", status="success", resource_type="cmdb_cost_allocation_rule", resource_id=rule_id, detail={"deleted_rule": snapshot}, ip_address=ip_address, user_agent=user_agent, ) await session.delete(rule) await session.commit() return snapshot # ─── Helpers for the CSV export ───────────────────────────────────── _CSV_HEADER = ( "date,integration_id,scope_id,resource_group,resource_id," "resource_name,service,region,usage_type,amount_usd," "amount_native,currency,owner_person_id,owner_department,cost_center," "tags_json" ) def _build_export_query( *, start: date, end: date, integration_id: str | None, dept_names: set[str] | None = None, limit: int, measure: str = "list", ): amount_col = fact_amount_expr(measure) # 保密红线:``measure=list`` 的导出(can_view_finance 受众含 # auditor / management,保密圈之外)每一行都必须单口径。USD 列 # 走 COALESCE(list, invoice) 过渡语义,native 列必须用同一条 # 回落规则 —— 此前 native 列裸读 ``amount_native``(实付口径), # 同一行同时携带 list USD 与实付 native,一相除折扣率就泄露。 # ``measure=invoice``(保密圈专属)保持实付 native 原样。 if measure == COST_MEASURE_INVOICE: native_col = CmdbCostFact.amount_native else: native_col = func.coalesce( CmdbCostFact.amount_native_list, CmdbCostFact.amount_native, ) where_clauses = [ CmdbCostFact.date >= start, CmdbCostFact.date <= end, ] if integration_id: where_clauses.append(CmdbCostFact.integration_id == integration_id) if dept_names is not None: where_clauses.append(CmdbCostFact.cost_owner_department.in_(dept_names)) return ( select( CmdbCostFact.date, CmdbCostFact.integration_id, CmdbCostFact.scope_id, CmdbCostFact.resource_group, CmdbCostFact.resource_id, CmdbCostFact.resource_name, CmdbCostFact.service, CmdbCostFact.region, CmdbCostFact.usage_type, amount_col.label("amount_usd"), native_col.label("amount_native"), CmdbCostFact.currency, CmdbCostFact.cost_owner_person_id, CmdbCostFact.cost_owner_department, CmdbCostFact.cost_center, # ``tags`` JSON — surfaced as one column with the JSON # payload. Finance opens the CSV in Excel and the JSON # cell is readable as-is; pivot via "split JSON" if # they want per-tag columns. Including the column in # the export closes the parity gap with the dashboard # surface so "who created this?" attribution is also # answerable from the bulk dump. CmdbCostFact.tags, ) .where(and_(*where_clauses)) .order_by(CmdbCostFact.date.asc()) .limit(limit) ) async def export_facts_csv( session: AsyncSession, *, start: date, end: date, integration_id: str | None = None, department: str | None = None, limit: int = 100_000, measure: str = "list", ) -> str: """Materialised CSV string of fact rows for offline analysis. Capped at ``limit`` rows (default 100k). The HTTP endpoint prefers ``stream_facts_csv`` to avoid building the full string in memory; this materialised variant is kept for tests + tooling. """ dept_names = await resolve_department_subtree_names(session, department) res = await session.execute(_build_export_query( start=start, end=end, integration_id=integration_id, dept_names=dept_names, limit=limit, measure=measure, )) lines = [_CSV_HEADER] for row in res: lines.append(",".join(_csv_cell(v) for v in row)) return "\n".join(lines) + "\n" async def stream_facts_csv( *, start: date, end: date, integration_id: str | None = None, department: str | None = None, limit: int = 100_000, measure: str = "list", ): """Yield CSV chunks for the ``/export.csv`` endpoint. Used with ``StreamingResponse`` so a 365-day × 100k-row dump doesn't sit fully in memory before the first byte hits the wire. Owns its own ``async_session_factory()`` rather than accepting the request-bound session for a deliberate reason: FastAPI's ``Depends(get_db)`` cleanup runs in a hand-off that's not guaranteed to keep the cursor live across the entire response body. Opening a fresh session here is robust and incurs only one extra connection acquisition for the export hot path. Yields the header row first, then one row per yield. The DB result-iterator is lazy on asyncmy so rows stream from TiDB without buffering the full set. """ from app.core.database import async_session_factory # UTF-8 BOM (```` encoded as ``b"\xef\xbb\xbf"``): without # it, Excel-on-Windows opens UTF-8 CSVs as Latin-1 / GBK and # mangles every Chinese resource name / RG name in the output. # Finance users opening the export expect "打开就能看" (looks # right on first open) without a manual import-via-数据-tab dance. # The BOM is also harmless on macOS / Linux Excel and on text # editors. ``charset=utf-8-sig`` on the response header is # informational; the BOM in the body is what actually toggles # Excel's autodetect behavior. yield ("" + _CSV_HEADER + "\n").encode("utf-8") rows_emitted = 0 async with async_session_factory() as session: dept_names = await resolve_department_subtree_names(session, department) res = await session.stream(_build_export_query( start=start, end=end, integration_id=integration_id, dept_names=dept_names, limit=limit, measure=measure, )) async for row in res: yield ( ",".join(_csv_cell(v) for v in row) + "\n" ).encode("utf-8") rows_emitted += 1 # Round-22 audit follow-up: when the hard ``limit`` was hit, surface # the truncation IN the file so finance can't accidentally treat # the truncated dump as the full set. The comment row starts with # ``#`` which is text-only (no formula trigger) and ignored by # most CSV ingest tools. Including the row count gives finance an # immediate sanity check. if rows_emitted >= limit: warn = ( f"# WARNING: result truncated at {limit} rows. The window " f"contained additional rows that were NOT included. Narrow " f"start/end or filter by integration_id to get the full set." ) yield (warn + "\n").encode("utf-8") # Also emit a trailer with the actual count so the operator can # reconcile streamed bytes against expected row count regardless # of truncation. yield ( f"# rows_emitted={rows_emitted} window={start}..{end}\n" ).encode() def _csv_cell(v: Any) -> str: """Escape a value for safe CSV output. Security: prepends a leading apostrophe to any cell whose first character is one of ``= + - @ tab CR``. Excel / Google Sheets / LibreOffice interpret these as the start of a formula (CSV formula injection) and will execute or auto-resolve them when a finance user opens the export. Adding the apostrophe forces the cell to render as literal text. The leading apostrophe is NOT a CSV special — it's a spreadsheet text-prefix convention. Standard CSV quoting rules (commas, newlines, double-quotes) are also applied below. JSON columns (dict / list) are serialised via ``json.dumps`` so finance opening the CSV sees standard JSON syntax (``{"Owner":"alice"}``), not Python repr (``{'Owner': 'alice'}``). ``ensure_ascii=False`` keeps Chinese tag values readable; the file already carries a UTF-8 BOM so Excel auto-detects encoding. """ if v is None: return "" if isinstance(v, (dict, list)): s = json.dumps(v, ensure_ascii=False, sort_keys=True) else: s = str(v) s = s.replace('"', '""') # Defang formula-injection vectors before standard quoting. if s and s[0] in ("=", "+", "-", "@", "\t", "\r"): s = "'" + s if "," in s or "\n" in s or '"' in s: return f'"{s}"' return s # ─── Cron orchestration: per-integration ingestion ────────────────── async def _audit_explicit_window_invocation( *, integration_id: str, start_date: date, end_date: date, ) -> None: """Service-layer invocation trace for explicit-window cost-facts runs. Prod forensics 2026-07-28: a 2,520-row / multi-month cost-facts reingest (AWS SG integration) landed at 06:46 UTC with NO ``audit_logs`` row and NO ``scheduled_job_runs`` row — something called ``sync_cost_facts_for_integration`` directly. The repo ships exactly such a path (``server/scripts/backfill_cost_facts.py``, which deliberately wrote no audit rows), and nothing stops future scripts / REPL sessions from doing the same. The HTTP ``/cmdb/cost/backfill`` endpoint and the MCP ``cost_backfill`` tool audit at their own layer, but any caller that imports the service function bypassed audit entirely. Fix: the SERVICE layer writes one ``actor_type="system"`` row per explicit-window invocation, committed in its OWN session BEFORE ingestion starts — so the trace survives even when the run crashes mid-ingest or the caller's session is rolled back. The cron default path (no window passed → rolling 3 days) never reaches this, so daily ticks don't spam ``audit_logs``; a recurring job running with a still-valid ``{start_date, end_date}`` backfill pin DOES trace daily, which is desirable — an active pin is a standing backfill. HTTP / MCP calls now produce two rows (their actor row + this system row); the redundancy is intentional — this row is the invariant "every explicit-window ingest is traceable", theirs carries the who. """ import inspect from app.audit.service import AuditService from app.core.database import async_session_factory # Best-effort caller attribution — first stack frame outside this # module. Enough to distinguish "scripts/backfill_cost_facts.py" # from "app/domains/cmdb/cost_router.py" from an ad-hoc REPL. caller: str | None = None try: stack = inspect.stack(context=0) try: for frame_info in stack[1:]: fname = frame_info.filename # Exact-match this module — a suffix check would also # swallow ``tests/.../test_cost_service.py`` frames. if fname == __file__: continue short = fname.rsplit("/server/", 1)[-1] caller = f"{short}:{frame_info.lineno} {frame_info.function}" break finally: del stack except Exception: # noqa: BLE001 — attribution must never block the trace caller = None async with async_session_factory() as audit_session: await AuditService(audit_session).log( trace_id=uuid.uuid4(), actor_type="system", action="cost.facts_sync.window_invoked", status="ok", resource_type="integration", resource_id=integration_id, detail={ "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "window_days": (end_date - start_date).days + 1, "caller": caller, "entrypoint": "sync_cost_facts_for_integration", }, ) await audit_session.commit() async def sync_cost_facts_for_integration( integration_id: str, *, start_date: date | None = None, end_date: date | None = None, ) -> dict[str, Any]: """Pull line-item cost facts from one cloud integration's driver and upsert into ``cmdb_cost_facts``. Default window: rolling 3 days ending yesterday (cron path). This matches ``sync_billing_daily_for_integration``'s default and is required for correctness: cloud billing exports keep ingesting late-arriving usage rows for ~2-3 days after the usage date (GCP BigQuery billing export back-fills usage rows and SUD/CUD credits; Azure Cost Management re-rates retroactively; AWS CUR re-publishes day partitions for ~24-48h). A single-day cron window captures only what was ingested by ~24h post-usage and leaves the rest in cmdb_cloud_billing_daily (which DOES re-fetch 3 days) — producing a systematic 5-12% undercount in cf vs bd on the last 2-3 days that converged to a $0 gap only on days older than the re-fetch window. UPSERT is idempotent (keyed by ``fact_key`` = date × integration × scope × resource × usage_type) so re-running for the same day overwrites stale values safely. Backfill: pass ``start_date`` / ``end_date`` explicitly via the scheduled-job config dict to cover wider gaps. For SSO clouds (Azure / AWS Org / Aliyun RD / GCP Org) the driver is called once per scope. For tenant-wide drivers (Tencent standalone IAM, Aliyun-as-tenant) the driver is called once with scope=None and the resulting facts cover everything that tenant owns. Returns a structured summary the scheduled_job_runs row stores in its ``details`` JSON column for the admin UI history. """ # UTC anchor — see forecast_month_end / detect_anomalies. Cron # default is "yesterday" relative to UTC so the same cron firing # on a UTC-deployed worker and a SH-deployed worker lands the # same window. from datetime import UTC from app.core.database import async_session_factory from app.domains.cmdb.lifecycle_service import _load_cloud_driver # Captured BEFORE default resolution: an explicit window is what # distinguishes a backfill (HTTP / MCP / cron pin / direct script # call) from the daily cron's rolling default. See # ``_audit_explicit_window_invocation`` for why this must trace at # the service layer. explicit_window = start_date is not None or end_date is not None end_date = end_date or (datetime.now(UTC).date() - timedelta(days=1)) # 3-day rolling window matches ``sync_billing_daily_for_integration`` # (lifecycle_service.py: ``start_dt = end_dt - timedelta(days=2)``). # See docstring above for the late-arrival rationale. start_date = start_date or (end_date - timedelta(days=2)) # Trace before the inverted-window guard: a bogus direct call # should leave a row too — "who tried" matters as much as "who ran". if explicit_window: await _audit_explicit_window_invocation( integration_id=integration_id, start_date=start_date, end_date=end_date, ) # Guard against an inverted window from a misconfigured manual # backfill / MCP backfill call. Silently looping zero times would # SKIP the pre-window DELETE (so phantom rows persist) AND still # record "status=ok upserted=0" against the run — exactly the # silent-success-without-progress shape that Round-37/38/39/40 # were trying to avoid. Fail loudly instead. if start_date > end_date: return { "status": "failed", "reason": ( f"invalid window: start_date={start_date} > " f"end_date={end_date}" ), "integration_id": integration_id, } async with async_session_factory() as session: integration = await session.get( Integration, uuid.UUID(integration_id), ) if integration is None or integration.kind != "cloud": return { "status": "skipped", "reason": "integration missing or not cloud", "integration_id": integration_id, } if not integration.enabled: return { "status": "skipped", "reason": "integration disabled", "integration_id": integration_id, } scope_lookup = await build_scope_lookup( session, integration_id=integration_id, ) # Load and reuse one driver instance across all scope calls # — driver __init__ does credential decrypt + token cache. try: driver = await _load_cloud_driver(session, integration) except Exception as exc: # noqa: BLE001 - explicit driver-load surface return { "status": "failed", "reason": ( f"driver load failed: {format_exception_for_log(exc)}" ), "integration_id": integration_id, } # Iterate scopes for SSO providers; fall back to tenant-wide # call when the provider has no scopes (standalone IAM-mode). # # Filter to tenant-level scope types only. Sub-tenant rows # (today: ``azure_resource_group`` minted by the Azure OpenAI # auto-grant flow) are NOT queryable units of the cloud cost # APIs — Azure Cost Management's POST query endpoint requires # a subscription GUID in the URL path, so passing an RG name # produces an ``InvalidSubscriptionId`` 400 and trips # ``partial_failure`` for the whole integration. RG-bound spend # is still attributed correctly: the parent subscription's # ingest emits per-resource facts whose ``resource_group`` # field matches an allocation rule (created alongside the RG # row by ``azure_openai_autogrant``) and the applicant is # stamped as cost owner. See PR fixing # ``cmdb.cost_facts_sync`` partial_failure caused by RG rows. scope_rows = (await session.execute( select(CmdbCloudScope) .where( CmdbCloudScope.integration_id == integration_id, CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == "active", CmdbCloudScope.scope_type.in_(TENANT_LEVEL_SCOPE_TYPES), ), )).scalars().all() from app.integrations.driver_base import CloudScopeRef targets: list[CloudScopeRef | None] = [] if scope_rows: for scope in scope_rows: targets.append(CloudScopeRef( scope_type=scope.scope_type, external_id=scope.external_id, display_name=scope.display_name, )) else: targets.append(None) # tenant-wide # Resolve CNY→USD once per ingestion run for CNY-native # drivers. Per-integration override on config_public takes # precedence; falls back to ``platform_settings`` value # refreshed weekly by ``ai_spending.cny_rate_refresh``. # Resolved here (not inside the driver) because the driver # has no DB session — its sync billing API call runs in a # worker thread via ``asyncio.to_thread``. Resolving once # also means a single rate stamps every fact in this run, # so an FX change mid-ingestion can't yield two USD numbers # for adjacent line items in the same batch. cny_per_usd: float | None = None # Direct attribute access — ``requires_cny_fx`` is declared on # ``CloudDriver`` so every concrete driver inherits it. Using # ``getattr(..., False)`` would silently mask a typo'd override # (e.g. ``require_cny_fx`` without the trailing ``s``) and the # CNY-native driver would then receive ``cny_per_usd=None`` and # either explode or silently land ``amount_usd=0`` rows. Same # bug class as the project memory ``feedback_no_getattr_default_ # for_model_columns`` — let the AttributeError be loud. if driver.requires_cny_fx: from app.core import fx_rate try: cny_per_usd = await fx_rate.resolve_cny_per_usd( session, integration_config=integration.config_public, ) except RuntimeError as exc: try: await driver.close() except Exception: # noqa: BLE001 pass return { "status": "failed", "reason": ( f"FX rate unavailable: {format_exception_for_log(exc)}" ), "integration_id": integration_id, } # Resolve the per-integration IAM-principal-tag opt-in once # per ingest run (not per scope). When set, the AWS driver # switches from per-resource granularity to per-(date × # service × tag-value) granularity — see # ``AWSCloudDriver._list_cost_facts_principal_tag_sync``. # Other drivers accept-and-ignore the kwarg. Looked up by the # integration slug: ``cost_facts.aws_principal_tag.`` # → tag key (e.g. ``"Owner"``). Empty / missing setting falls # through to the default per-resource path. AWS-only today; # if Aliyun / Tencent ship an equivalent dimension we'd # promote the setting key family then. principal_tag_keys: list[str] | None = None if integration.provider == "aws": slug = integration.slug or str(integration.id) setting_key = f"cost_facts.aws_principal_tag.{slug}" row = await session.get(PlatformSetting, setting_key) if row is not None and row.value: raw = row.value # platform_settings.value is JSON — accept a string # ("Owner") or a list (["Owner", "Project"]). Anything # else is ignored loudly so an admin typo doesn't # silently swap modes. if isinstance(raw, str): keys_clean = [raw.strip()] if raw.strip() else [] elif isinstance(raw, list): keys_clean = [ str(k).strip() for k in raw if str(k).strip() ] else: logger.warning( "cost_facts: ignored %s — value must be string " "or list of strings, got %r", setting_key, type(raw).__name__, ) keys_clean = [] if keys_clean: principal_tag_keys = keys_clean # Resolve the BigQuery CUR source once per ingest run (not per # scope). When the AWS integration is configured with # ``billing_source=bigquery``, the cost-facts cron must read # from the CUR-in-BQ path rather than Cost Explorer (whose # creds prod doesn't have). Other providers / AWS-on-CE return # None here and the kwarg is a no-op. Raises loudly if the # config is half-set — see ``_resolve_aws_bq_billing``. from app.domains.cmdb.lifecycle_service import ( _resolve_aws_bq_billing, ) try: bq_billing = await _resolve_aws_bq_billing(session, integration) except RuntimeError as exc: try: await driver.close() except Exception: # noqa: BLE001 pass return { "status": "failed", "reason": ( f"BQ billing source misconfigured: " f"{format_exception_for_log(exc)}" ), "integration_id": integration_id, } # AWS CUR-in-BigQuery (per_month / CUR 2.0): hand the driver the # full target scope set so its cost-facts path scans all linked # accounts in ONE query per window instead of one query per # account. The CUR month-tables are DAY-partitioned and NOT # clustered by account, so the per-account ``WHERE # line_item_usage_account_id = ?`` filter bills identical bytes # whether it matches one account or all — making the per-account # fan-out a pure N× scan-cost multiplier (prod 2026-06: ~38×, # ~$2.3k/yr steady-state bleed + multi-hundred-$ backfill spikes, # billed to the GCP project running the SA's BQ jobs). Set only # when ``bq_billing`` is present — that is AWS-BQ mode, so the # driver is always the AWS driver which owns this attribute; the # CE path and other providers leave it None and keep the per- # account streaming path. The driver guards activation further # (per_month + per_product grain + bounded window span) so an # unbounded backfill can't hold every account's facts at once. if bq_billing is not None: driver._cost_facts_all_account_scopes = [ t for t in targets if t is not None ] total_upserted = 0 per_scope: list[dict[str, Any]] = [] errors: list[str] = [] # Per-scope timeout — defends against a single hung scope # (BQ row-iterator stall, network hiccup, oversized result # set) blocking the rest of the cron. Without this guard the # whole sync wedges in ``running`` state until manually # intervened (2026-05-19 prod incident: a single account's # CUR 2.0 query produced ~66k grouped rows after the new # per_resource path, exhausting upsert throughput at ~30s # per scope when prior CUR 1.0 path averaged ~3s). # # Module-level constant ``_COST_FACTS_PER_SCOPE_TIMEOUT_SEC`` # is patchable for tests (we shrink to ~0.2s there to keep # the suite fast). The 300-second prod default is generous # enough for any single AWS account's worst-case row volume # (largest prod scope's last successful CUR 1.0 run finished # in <2 min) while keeping a 60-scope cron tick bounded by # ``60 × 5min = 5h`` in absolute worst case (in practice # most scopes finish in seconds; only the rare runaway # hits the cap). per_scope_timeout_sec = _COST_FACTS_PER_SCOPE_TIMEOUT_SEC try: for scope_ref in targets: scope_label = ( scope_ref.external_id if scope_ref else "" ) try: facts = await asyncio.wait_for( driver.list_cost_facts( start_date, end_date, scope=scope_ref, cny_per_usd=cny_per_usd, principal_tag_keys=principal_tag_keys, bq_billing=bq_billing, ), timeout=per_scope_timeout_sec, ) except TimeoutError: msg = ( f"scope {scope_label} exceeded " f"{per_scope_timeout_sec}s timeout; advancing" ) errors.append(msg) per_scope.append({ "scope": ( scope_ref.external_id if scope_ref else None ), "status": "timeout", "error": msg, "facts": 0, }) continue except NotImplementedError: per_scope.append({ "scope": ( scope_ref.external_id if scope_ref else None ), "status": "not_supported", "facts": 0, }) continue except Exception as exc: # noqa: BLE001 - driver surface # Use the canonical ``format_exception_for_log`` so a # transport-level exception whose ``str(exc)`` is # empty (httpx.ReadTimeout / RemoteProtocolError / # some ConnectError paths) AND an AppException-derived # raise (whose ``str()`` is also empty by design) # both surface the exception class + message in the # failure-notifier line. Without this canonical # helper, a single bad scope used to produce # ``: `` with no clue what failed — exactly # the prod cost_facts_sync incident on 2026-05-21 # (Azure subscription 7bc8ab17-... single-scope # transient that left ops blind to the exception # type). rendered = format_exception_for_log(exc, max_len=300) errors.append(f"{scope_label}: {rendered}") per_scope.append({ "scope": ( scope_ref.external_id if scope_ref else None ), "status": "failed", "error": rendered, "facts": 0, }) continue # Stale-fact eviction before re-UPSERT (Round-36, 2026-05-18). # The driver's grouped SUM emits one fact per natural-key # tuple (date × scope × resource × usage_type) with # ``if native == 0.0: continue`` dropping groups whose net # cost is exactly zero — typical when SP CoveredUsage and # SP Negation pair up after AWS publishes the negation row # (often 12-36h after the covered-usage row). Without a # pre-window DELETE, the earlier run's emission of the # not-yet-cancelled SP-positive group survives forever: # the later run finds that fact_key sums to $0, skips # emission, and UPSERT never overwrites the stale row. # Same shape will bite any provider whose natural-key set # shrinks across runs (a usage_type disappears, a # resource_id is reclassified, etc.). Bit # ``pingcap-china-outbound`` (+$165 over 16 days) and # ~10 other AWS scopes in 2026-05-18 audit (clinic-dev # +$664, dbaas-prod-clinic +$562, etc.). The DELETE # joins the same SQLAlchemy session as ``ingest_cost_facts`` # below, so the first ingest-batch commit makes the # DELETE durable atomically with the first written batch. # On the empty-facts branch we explicitly commit so the # eviction lands even when the scope's window legitimately # went to zero cost. # Chunk the DELETE per-day AND LIMIT-N within each day # (Round-39, 2026-05-18). Round-38 (PR #381, migration 181) # added the ``(scope_id, date)`` composite index and # bet that an index-pinpoint plan alone would keep # working-set under the 1 GiB ``tidb_mem_quota_query`` # cap — but the AWS ``PingCAP`` payer scope hit prod # 8175 again on 2026-05-18 because ~45k–50k rows/day × # ~16 KiB ``raw`` JSON payload is ~720–800 MiB, right # at the edge of the cap, and a single transactional # plan still has to materialize every row's payload # to satisfy the DELETE. The index gives TiDB a tight # plan (no over-scan) but does NOT bound the per- # statement row-payload size. ``DELETE … LIMIT N`` # does — it caps the statement's working set strictly # to N rows. Loop until ``rowcount < N`` is the # standard idiom for "delete everything matching, in # bounded chunks". TiDB supports ``DELETE … LIMIT``; # SQLAlchemy's typed ``delete()`` builder does not, # so this branch drops to a parameterised ``text()`` # — safe because the only interpolated values come # via bind parameters (no SQL injection surface). # Per-day outer loop is still required: a full-window # DELETE on a multi-year scope would still queue a # multi-GiB row payload before the LIMIT trims it. scope_db_id: str | None = None if scope_ref is not None and scope_ref.external_id: scope_db_id = scope_lookup.get(scope_ref.external_id) # Eviction unit: per (scope, date) for a scope-addressed # call, per (integration, date) for a tenant-wide one. # # A tenant-wide call is reached only when the integration # has NO active tenant-level ``cmdb_cloud_scopes`` row, so # ``targets == [None]`` and the single driver call returns # the WHOLE integration's window — which makes # (integration, date) the correct authority, and makes the # per-scope worry (scope A's DELETE eating scope B's rows) # structurally impossible: there is no second iteration. # Before this branch existed the gate skipped eviction # entirely for that path, so such an integration's facts # were never evicted at all: a line that later netted to # zero upstream survived forever, AND the pre-2026-07-29 # non-``usage`` rows (plain fact_key) were never removed # when the new bucket-bearing key landed as a second row — # a permanent double-count of credit / adjustment / tax in # the finance waterfall and all chargeback. Facts whose # scope did not resolve to a local row cannot appear on the # scope-addressed path: ``targets`` is built from the same # ``cmdb_cloud_scopes`` read as ``scope_lookup`` and with a # STRICTER filter (active + tenant-level), so # ``scope_lookup.get`` cannot miss for a ``scope_ref``. # Exposure parity with the per-scope branch: the window is # deleted before ``ingest_cost_facts`` runs, so an ingest # that raises mid-window leaves the deleted days empty # until the next tick re-ingests them — accepted for a # scope, and accepted here for the tenant, on the same # rolling-window rationale. tenant_wide_evict = scope_db_id is None and scope_ref is None if scope_db_id or tenant_wide_evict: from sqlalchemy import text as _sql_text one_day = timedelta(days=1) cur_date = start_date # 1000 rows/chunk caps per-statement working set; # per-day commit caps cross-chunk transaction # memtable. Round-39 (PR #383) restored the # per-chunk LIMIT but left every chunk in one # txn — on the AWS ``PingCAP`` payer scope's # 16-day backfill, ~720 chunks × undo+lock # entries per row accumulated past 1 GiB # ``tidb_mem_quota_query`` and still 8175'd on # ``date=2026-05-12``. Per-day commit bounds the # txn's memtable to one day's eviction (~45k # rows on the heaviest scope), independent of # how many days the cron / backfill spans. # Atomicity-wise: a day's eviction is the # natural logical unit (one date is what # downstream UPSERT replaces); committing per- # day means a mid-window crash leaves earlier # days fully evicted but the failed day in its # pre-eviction state, which the next cron tick # naturally repairs (idempotent ON DUPLICATE # KEY UPDATE). delete_chunk = 1000 # Both shapes stay on an index-range plan: # ``ix_facts_scope_date`` (scope_id, date) for the # per-scope form, ``ix_facts_date_int`` # (date, integration_id) for the tenant-wide one. delete_stmt = _sql_text( "DELETE FROM cmdb_cost_facts " "WHERE integration_id = :iid " + ("" if tenant_wide_evict else "AND scope_id = :sid ") + "AND `date` = :d " "LIMIT :n" ) while cur_date <= end_date: params: dict[str, Any] = { "iid": integration_id, "d": cur_date, "n": delete_chunk, } if not tenant_wide_evict: params["sid"] = scope_db_id while True: res = await session.execute(delete_stmt, params) if (res.rowcount or 0) < delete_chunk: break await session.commit() cur_date = cur_date + one_day if not facts: if scope_db_id or tenant_wide_evict: await session.commit() per_scope.append({ "scope": ( scope_ref.external_id if scope_ref else None ), "status": "ok", "facts": 0, }) continue summary = await ingest_cost_facts( session, integration_id=integration_id, facts=facts, scope_lookup=scope_lookup, ) total_upserted += summary["upserted"] per_scope.append({ "scope": ( scope_ref.external_id if scope_ref else None ), "status": "ok", "facts": summary["upserted"], }) finally: try: await driver.close() except Exception as close_exc: # noqa: BLE001 # Don't let a hung connection pool / late-cancelled # socket close obscure the actual sync result — the # caller already has ``per_scope`` / ``errors`` # populated by the time we reach ``finally``. But DO # log it: a recurring close failure points to a real # driver-level resource leak (e.g. httpx AsyncClient # not exiting cleanly, vendor SDK background thread # stuck), which is otherwise invisible to ops. Same # ``logger.exception`` posture as the earlier RuntimeError # branches at L2951 / L3014. logger.exception( "cost_facts_sync driver.close() raised for integration " "%s: %s", integration_id, close_exc, ) result: dict[str, Any] = { "status": ( "ok" if not errors else "partial_failure" if total_upserted > 0 else "failed" ), "integration_id": integration_id, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "upserted": total_upserted, "per_scope": per_scope, } if errors: result["errors"] = errors[:10] return result # ── Bedrock per-profile (UsedBy tag) breakdown ─────────────────────── # # Isolated, additive sibling of the cost-facts ingest above. Writes the # secondary "profile" dimension (the ``UsedBy`` IAM-principal tag — e.g. # ``investibook-dev`` / ``aiops`` — that subdivides one owner's Bedrock # spend) into ``cmdb_bedrock_profile_facts``, NOT ``cmdb_cost_facts``, # so it can never perturb the totals/by-model/dashboard aggregations # over that delicate table. Driven by the same cron window as the main # cost-facts sync (see the cron handler) so the per-AK detail page's # by-profile sum reconciles to its ``mtd_usd``. def _profile_fact_key( *, integration_id: str, fact_date: date, scope_external_id: str | None, owner_email: str, profile_tag_key: str, profile: str, service: str, sub_profile: str | None = None, ) -> str: """sha256 hex of the profile fact's natural key (stable across re-ingestion). Mirrors :func:`compute_fact_key`'s NUL-marker / delimiter convention so two no-scope rows can't collide. ``profile_tag_key`` joined the natural key in migration 435 (one table now holds several tag axes — ``UsedBy`` workloads AND the per-API-Key ``iamPrincipal/helix-request-number`` axis); without it two axes' rows for the same (day, owner, service) would collide and the second would overwrite the first. ``sub_profile`` joined in migration 452 — the per-API-Key axis now breaks each key down by ``UsedBy`` too, so a per-key row is (date × key × UsedBy × service). Without it the (date × key × service) tuples across different UsedBy values would collide. NULL (every pre-452 row, and the UsedBy workload axis) uses the NUL marker component. Inserting the field DOES shift the hash of a legacy NULL-sub row, but that is harmless: the ingest's per-(scope, date) window DELETE drops the old rows before re-insert, so any re-synced window reconverges on the new keys with no duplicates, and windows never re-synced simply keep their old (still-unique) keys.""" def _component(v: str | None) -> str: return ( _FACT_KEY_NULL_MARKER if v is None else v.replace("\x01", "\\x01") ) parts = [ fact_date.isoformat(), str(integration_id), _component(scope_external_id), _component(owner_email), _component(profile_tag_key), _component(profile), _component(sub_profile), _component(service), ] return hashlib.sha256("\x01".join(parts).encode("utf-8")).hexdigest() async def _resolve_aws_setting_first_key( session: AsyncSession, setting_key: str, ) -> str | None: """Read a ``platform_settings`` JSON value that may be a string ("Owner") or a list (["Owner", ...]) and return the first non-empty key, or None. Matches the parsing the main cost-facts sync does for ``cost_facts.aws_principal_tag.``.""" row = await session.get(PlatformSetting, setting_key) if row is None or not row.value: return None raw = row.value if isinstance(raw, str): return raw.strip() or None if isinstance(raw, list): for k in raw: if str(k).strip(): return str(k).strip() return None async def _resolve_aws_setting_keys( session: AsyncSession, setting_key: str, ) -> list[str]: """Every non-empty key in a ``platform_settings`` value that may be a string ("UsedBy") or a list (["UsedBy", "iamPrincipal/helix-…"]). :func:`_resolve_aws_setting_first_key` keeps the single-axis contract for the owner tag (Cost Explorer's GroupBy has exactly one slot for it). The PROFILE axis, in contrast, may legitimately be pulled several times — once per tag key — because each pull is an independent full breakdown of the same owner total (migration 435). Order is preserved and duplicates collapse. """ row = await session.get(PlatformSetting, setting_key) if row is None or not row.value: return [] raw = row.value values = [raw] if isinstance(raw, str) else ( list(raw) if isinstance(raw, list) else [] ) out: list[str] = [] for k in values: cleaned = str(k).strip() if cleaned and cleaned not in out: out.append(cleaned) return out async def sync_bedrock_profile_facts_for_integration( integration_id: str, *, start_date: date | None = None, end_date: date | None = None, ) -> dict[str, Any]: """Pull the Bedrock per-(owner × profile × service) breakdown for one AWS integration and upsert into ``cmdb_bedrock_profile_facts``. No-op (``status=skipped``) unless the integration is AWS and BOTH settings are configured: * ``cost_facts.aws_principal_tag.`` — the owner tag (the same one the main cost-facts sync attributes Bedrock spend on), AND * ``cost_facts.aws_profile_tag.`` — the secondary profile tag to break each owner's spend down by (e.g. ``UsedBy``). May be a LIST of tag keys (migration 435): each is pulled as its own independent breakdown axis and stored with its ``profile_tag_key``. Configure ``["UsedBy", "iamPrincipal/helix-request-number"]`` to get both the workload axis and the per-API-Key axis (the latter is what gives a key pinned to $0 by the shared-Owner rule its own number). Default window matches ``sync_cost_facts_for_integration`` (rolling 3 days ending yesterday, UTC) so both tables move on the same watermark. Idempotent: per-(scope, date) pre-window DELETE then fact_key-keyed upsert, so re-runs converge. """ from app.core.database import async_session_factory from app.domains.cmdb.lifecycle_service import _load_cloud_driver end_date = end_date or (datetime.now(UTC).date() - timedelta(days=1)) start_date = start_date or (end_date - timedelta(days=2)) if start_date > end_date: return { "status": "failed", "reason": ( f"invalid window: start_date={start_date} > " f"end_date={end_date}" ), "integration_id": integration_id, } async with async_session_factory() as session: integration = await session.get( Integration, uuid.UUID(integration_id), ) if ( integration is None or integration.kind != "cloud" or integration.provider != "aws" ): return { "status": "skipped", "reason": "integration missing / not an AWS cloud", "integration_id": integration_id, } if not integration.enabled: return { "status": "skipped", "reason": "integration disabled", "integration_id": integration_id, } slug = integration.slug or str(integration.id) owner_tag_key = await _resolve_aws_setting_first_key( session, f"cost_facts.aws_principal_tag.{slug}", ) # Several profile axes may be configured (migration 435): the # ``UsedBy`` workload axis plus the per-API-Key # ``iamPrincipal/helix-request-number`` axis. Each is pulled and # stored independently; they are NOT summed together. profile_tag_keys = await _resolve_aws_setting_keys( session, f"cost_facts.aws_profile_tag.{slug}", ) if not owner_tag_key or not profile_tag_keys: return { "status": "skipped", "reason": ( "no aws_principal_tag / aws_profile_tag configured" ), "integration_id": integration_id, } # The workload tag the per-API-Key axis is sub-divided by # (migration 452): the first configured axis that ISN'T itself a # per-key principal tag — i.e. ``UsedBy`` in the canonical # ``["UsedBy", "iamPrincipal/helix-request-number"]`` config — # falling back to the default when the list names only per-key # axes. sub_profile_tag = next( ( k for k in profile_tag_keys if k.strip().lower() not in _PER_KEY_PRINCIPAL_AXES ), _DEFAULT_SUB_PROFILE_TAG, ) # Profile breakdown comes from Cost Explorer's principal-tag # GroupBy; an integration on the BigQuery CUR billing source has # no per-call CE creds. Skip rather than half-run. A HALF-set BQ # config raises RuntimeError — mirror the main sync and treat that # as skipped too (do NOT fall through to CE, which would hammer # creds that may lack CE access and emit noisy AccessDenied). from app.domains.cmdb.lifecycle_service import _resolve_aws_bq_billing try: bq_billing = await _resolve_aws_bq_billing(session, integration) except RuntimeError as exc: return { "status": "skipped", "reason": ( f"BQ billing source misconfigured: " f"{format_exception_for_log(exc)}" ), "integration_id": integration_id, } if bq_billing is not None: return { "status": "skipped", "reason": "billing_source=bigquery has no CE principal tags", "integration_id": integration_id, } try: driver = await _load_cloud_driver(session, integration) except Exception as exc: # noqa: BLE001 return { "status": "failed", "reason": ( f"driver load failed: {format_exception_for_log(exc)}" ), "integration_id": integration_id, } async def _close_driver() -> None: try: await driver.close() except Exception: # noqa: BLE001 pass if not hasattr(driver, "list_bedrock_profile_facts"): await _close_driver() return { "status": "skipped", "reason": "driver has no list_bedrock_profile_facts", "integration_id": integration_id, } # The owner cost-allocation tag is an IAM-PRINCIPAL tag, so the # CE fan-out returns EVERY service the owner touched (EC2/S3/…), # not just Bedrock. Restrict to the SAME AWS AI-whitelist patterns # the per-AK ``mtd_usd`` uses (``_sum_with_service_patterns`` → # ``service ILIKE '%%'``) so the per-profile sum # reconciles to ``mtd_usd`` by construction instead of inflating # it with the principal's non-Bedrock spend. from app.domains.ai_spending.whitelist import load_whitelist try: wl = await load_whitelist(session) except Exception: # Missing/invalid whitelist setting — don't leak the driver. await _close_driver() raise service_patterns = [ p for p in (wl.get("aws") or []) if p and p.strip() ] if not service_patterns: await _close_driver() return { "status": "skipped", "reason": "aws AI service whitelist not configured", "integration_id": integration_id, } _patterns_lower = [p.lower() for p in service_patterns] def _is_bedrock_service(svc: str | None) -> bool: s = (svc or "").lower() return any(p in s for p in _patterns_lower) # Resolve CNY→USD once, the same way the main cost-facts sync does # (only CNY-native drivers need it; AWS is USD so this stays None). cny_per_usd: float | None = None # Direct attribute access — ``requires_cny_fx`` is declared on # ``CloudDriver`` so every concrete driver inherits it. A # ``getattr(..., False)`` default would silently mask a typo'd # override and land ``amount_usd=0`` CNY rows; match the loud # attribute access used by the primary ingest path above. if driver.requires_cny_fx: from app.core import fx_rate try: cny_per_usd = await fx_rate.resolve_cny_per_usd( session, integration_config=integration.config_public, ) except RuntimeError as exc: try: await driver.close() except Exception: # noqa: BLE001 pass return { "status": "failed", "reason": ( f"FX rate unavailable: {format_exception_for_log(exc)}" ), "integration_id": integration_id, } scope_rows = (await session.execute( select(CmdbCloudScope).where( CmdbCloudScope.integration_id == integration_id, CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == "active", CmdbCloudScope.scope_type.in_(TENANT_LEVEL_SCOPE_TYPES), ), )).scalars().all() from app.integrations.driver_base import CloudScopeRef total_upserted = 0 per_scope: list[dict[str, Any]] = [] errors: list[str] = [] try: for scope in scope_rows: ext = scope.external_id if not (ext and str(ext).isdigit() and len(str(ext)) == 12): # Profile breakdown is Bedrock-only and Bedrock lives # on 12-digit linked accounts; skip non-account scopes. continue scope_ref = CloudScopeRef( scope_type=scope.scope_type, external_id=ext, display_name=scope.display_name, ) async def _fetch_scope_facts( scope_ref: CloudScopeRef = scope_ref, ) -> list[dict[str, Any]]: # One CE fan-out per configured axis. Only the vendor # fetches live here (under the per-scope timeout); the # DB ingest runs OUTSIDE so a timeout can never cancel # a coroutine mid-flush (same split the cost-facts # per-scope timeout keeps). ``to_thread`` means a # cancel here leaks the boto3 thread but leaves the # shared session untouched. facts: list[dict[str, Any]] = [] for profile_tag_key in profile_tag_keys: is_per_key = ( profile_tag_key.strip().lower() in _PER_KEY_PRINCIPAL_AXES ) if is_per_key and hasattr( driver, "list_bedrock_per_key_profile_facts", ): # Per-API-Key axis crossed with UsedBy (452): # fixes the key via Filter and groups by # (UsedBy × SERVICE) so each key gets its OWN # workload split. Rows carry ``sub_profile``. axis_facts = ( await driver .list_bedrock_per_key_profile_facts( start_date, end_date, scope=scope_ref, owner_tag_key=owner_tag_key, key_tag_key=profile_tag_key, sub_tag_key=sub_profile_tag, cny_per_usd=cny_per_usd, ) ) else: # Workload axis (``UsedBy``) — owner fan-out, # grouped by (tag × SERVICE); ``sub_profile`` # absent. Also the fallback for a driver too # old to know the cross method. axis_facts = ( await driver.list_bedrock_profile_facts( start_date, end_date, scope=scope_ref, owner_tag_key=owner_tag_key, profile_tag_key=profile_tag_key, cny_per_usd=cny_per_usd, ) ) for f in axis_facts: # Restrict to the Bedrock service set (same # whitelist as mtd_usd) and stamp the axis. if not _is_bedrock_service(f.get("service")): continue f["profile_tag_key"] = profile_tag_key facts.append(f) return facts try: # Bound the vendor fetches so a wedged CE call can't # hold the shared cost-facts lock to its 900s TTL and # let the next cron tick race this window's DELETE. facts = await asyncio.wait_for( _fetch_scope_facts(), timeout=_BEDROCK_PROFILE_PER_SCOPE_TIMEOUT_SEC, ) # Ingest inside the per-scope try so a single bad scope # (driver dict shape, DB error) is recorded and skipped # without aborting the remaining scopes. upserted = await ingest_bedrock_profile_facts( session, integration_id=integration_id, scope_external_id=str(ext), start_date=start_date, end_date=end_date, facts=facts, ) except ( NotImplementedError, RuntimeError, KeyError, TimeoutError, ) as exc: # Discard any uncommitted work from the failed scope — # notably ingest's per-window DELETE, which runs before # the row build. Without this rollback the NEXT scope's # commit would flush that stranded DELETE and wipe this # scope's window with no replacement. try: await session.rollback() except Exception: # noqa: BLE001 pass rendered = ( f"per-scope timeout after " f"{_BEDROCK_PROFILE_PER_SCOPE_TIMEOUT_SEC:.0f}s" if isinstance(exc, TimeoutError) else format_exception_for_log(exc, max_len=300) ) errors.append(f"{ext}: {rendered}") per_scope.append({ "scope": ext, "status": "failed", "error": rendered, "facts": 0, }) continue total_upserted += upserted per_scope.append({ "scope": ext, "status": "ok", "facts": upserted, }) finally: try: await driver.close() except Exception: # noqa: BLE001 pass status = "ok" if errors and total_upserted == 0: status = "failed" elif errors: status = "partial_failure" result: dict[str, Any] = { "status": status, "integration_id": integration_id, "window": { "start": start_date.isoformat(), "end": end_date.isoformat(), }, "upserted": total_upserted, "per_scope": per_scope, } if errors: result["errors"] = errors[:10] return result async def ingest_bedrock_profile_facts( session: AsyncSession, *, integration_id: str, scope_external_id: str, start_date: date, end_date: date, facts: list[dict[str, Any]], ) -> int: """Idempotently replace one scope's profile facts for the window. Pre-window DELETE by (integration, scope, date) — the table is tiny (owners × profiles × services × days), so a single full-window DELETE is well within memory bounds (no chunking needed, unlike the 16 KiB-``raw`` cmdb_cost_facts path) — then ``INSERT ... ON DUPLICATE KEY UPDATE`` keyed by ``fact_key``. The DELETE makes the natural-key set shrink-safe: a profile that stops appearing across runs (a workload retired, a CE re-rate that zeroes a cell) is evicted rather than stranded as a stale row. """ from sqlalchemy import delete as _sql_delete from sqlalchemy.dialects.mysql import insert as _mysql_insert await session.execute( _sql_delete(CmdbBedrockProfileFact).where( CmdbBedrockProfileFact.integration_id == integration_id, CmdbBedrockProfileFact.scope_external_id == scope_external_id, CmdbBedrockProfileFact.date >= start_date, CmdbBedrockProfileFact.date <= end_date, ) ) if not facts: await session.commit() return 0 now = datetime.now(tz=UTC) rows: list[dict[str, Any]] = [] for f in facts: fact_date = f["date"] owner = str(f.get("owner_email") or "") profile = str(f.get("profile") or "(unallocated)") service = str(f.get("service") or "Other") # Second breakdown dimension on the per-API-Key axis (migration # 452). Absent / None for the UsedBy workload axis and every # pre-452 driver; kept out of the fact_key's empty-string bucket # (it uses the NUL marker) so a genuine "(unallocated)" UsedBy # value stays distinct from "no sub axis at all". sub_raw = f.get("sub_profile") sub_profile = ( str(sub_raw)[:256] if sub_raw is not None else None ) # Default keeps pre-435 callers (and any driver that predates the # stamp) on the axis this table shipped with. tag_key = str(f.get("profile_tag_key") or "UsedBy")[:128] rows.append({ "fact_key": _profile_fact_key( integration_id=integration_id, fact_date=fact_date, scope_external_id=scope_external_id, owner_email=owner, profile_tag_key=tag_key, profile=profile, service=service, sub_profile=sub_profile, ), "date": fact_date, "integration_id": integration_id, "scope_external_id": scope_external_id, "owner_email": owner, "profile_tag_key": tag_key, "profile": profile, "sub_profile": sub_profile, "service": service, "amount_usd": _q6(f.get("amount_usd")), "amount_native": ( _q6(f["amount_native"]) if f.get("amount_native") is not None else None ), "currency": str(f.get("currency") or "USD")[:8], "usage_quantity": ( _q6(f["usage_quantity"]) if f.get("usage_quantity") is not None else None ), "usage_unit": f.get("usage_unit"), "fetched_at": now, }) # Dedup within the batch on fact_key (defensive — CE shouldn't emit # two groups for the same natural key, but ON DUPLICATE KEY UPDATE on # a multi-row INSERT with in-batch dups is engine-defined; collapse # by summing so we never silently drop a cell). merged: dict[str, dict[str, Any]] = {} for r in rows: key = r["fact_key"] if key in merged: prev = merged[key] prev["amount_usd"] = (prev["amount_usd"] or 0) + (r["amount_usd"] or 0) if r["amount_native"] is not None: prev["amount_native"] = ( (prev["amount_native"] or 0) + r["amount_native"] ) if r["usage_quantity"] is not None: prev["usage_quantity"] = ( (prev["usage_quantity"] or 0) + r["usage_quantity"] ) else: merged[key] = dict(r) stmt = _mysql_insert(CmdbBedrockProfileFact).values(list(merged.values())) stmt = stmt.on_duplicate_key_update( amount_usd=stmt.inserted.amount_usd, amount_native=stmt.inserted.amount_native, currency=stmt.inserted.currency, usage_quantity=stmt.inserted.usage_quantity, usage_unit=stmt.inserted.usage_unit, fetched_at=stmt.inserted.fetched_at, ) await session.execute(stmt) await session.commit() return len(merged) # Re-exports for convenience. # ── ``raw`` retention (storage optimisation, 2026-09-01) ───────── #: Days of ``cmdb_cost_facts.raw`` kept online. ``raw`` is the vendor's #: verbatim billing line item, stored purely for post-hoc "why is this #: number what it is" spelunking — it has **no reader anywhere in the #: codebase** (the only reference is the ingest write below). Measured #: on prod 2026-09-01: avg 311 bytes/row × ~1.0M rows/month ≈ 0.33 GB #: per month, ~3.6 GB across the 11 months retained, against a #: 10.5 GB table. Ninety days keeps every window an operator has ever #: actually needed to debug (the widest re-rating window any vendor #: uses is ~45 days) and drops the rest. #: #: Admin-tunable via ``platform_settings['cost_facts.raw_retention_days']``. _RAW_RETENTION_DAYS_DEFAULT = 90 #: Floor on the tunable. Below the widest vendor re-rating window a purge #: would start eating ``raw`` for days that are still being re-published, #: which is the one period where an operator plausibly needs it. _RAW_RETENTION_DAYS_MIN = 45 #: How far the purge has swept, as an ISO date string. Days at or below #: this are known-purged and are never re-scanned — without it every #: nightly tick would range-scan ~9M index entries to find nothing. _RAW_PURGE_WATERMARK_KEY = "cost_facts.raw_purged_through_date" _RAW_RETENTION_SETTING_KEY = "cost_facts.raw_retention_days" async def resolve_raw_retention_days(session: AsyncSession) -> int: """Resolve the ``raw`` retention window (DB-first, clamped). Mirrors ``get_sweep_drop_tolerance_percent``'s DB-first shape so an admin can widen the window without a deploy. Garbage values fall back to the default rather than raising — this feeds a cron whose failure mode should be "kept more data than needed", never "the tick died". """ row = await session.get(PlatformSetting, _RAW_RETENTION_SETTING_KEY) raw_value = row.value if row is not None else None try: days = int(raw_value) # type: ignore[arg-type] except (TypeError, ValueError): days = _RAW_RETENTION_DAYS_DEFAULT return max(_RAW_RETENTION_DAYS_MIN, min(days, 3650)) async def purge_cost_fact_raw( session: AsyncSession, *, retention_days: int | None = None, batch_size: int = 2000, max_days_per_run: int = 30, max_seconds_per_run: float = 600.0, ) -> dict[str, Any]: """NULL out ``cmdb_cost_facts.raw`` for accounting days older than the retention window. Walks **one accounting day at a time** rather than issuing a single ``UPDATE ... WHERE date < cutoff``. Two reasons, both learned the hard way on this table: * TiDB bills a write transaction's membuffer to the session and kills whatever statement is running when the instance-level cap trips (error 8176). A single UPDATE spanning ~9M rows is exactly the shape that trips it. Per-day + per-batch commits keep each transaction small and short — the latter matters independently, because a long-running transaction pins the cluster GC safepoint (``tidb_gc_life_time`` is 10m on this cluster). * ``date`` is the leading column of ``ix_facts_date_id``, so ``WHERE date = :day`` is an index range scan over that one day (~30k rows) instead of a table scan. The watermark in ``platform_settings`` means a steady-state tick advances exactly one day and touches one day's index range. Days below the watermark are never revisited; the ingest-side guard in ``ingest_cost_facts`` is what keeps a late backfill from re-populating ``raw`` underneath it. **Interruptible by construction (2026-09-02).** The first production run was killed by the scheduler's 1800s hard timeout having purged zero rows — the loop itself is fast (measured in-region at ~6,600 rows/s, so the 30-day first slice is ~85s of work), and the worker was healthy throughout, so the tick was starved rather than slow. What made that fatal was bookkeeping, not speed: the watermark used to be written once after the loop, so a killed run threw away everything it had done. Worse, ``days_processed`` counts days whether or not they had rows, so the next run would start at ``min(date)`` again, spend its whole day budget re-walking already-empty days, and **never reach new work** — a job that fails once could never recover on its own. Two changes make the tick safe to kill at any instant: * the watermark is committed **after every completed day**, so progress is never larger than one day's worth of loss; and * ``max_seconds_per_run`` is a soft budget checked between days. Reaching it is a normal, successful outcome — the run stops, keeps its watermark, and reports ``stopped_early``. The budget must stay comfortably below the scheduler's hard timeout: the hard timeout is a kill, this is a clean stop, and a clean stop is what keeps the summary honest and the next tick productive. Returns a summary dict for the ``scheduled_job_runs`` row. """ if retention_days is None: retention_days = await resolve_raw_retention_days(session) cutoff = datetime.now(UTC).date() - timedelta(days=retention_days) watermark_row = await session.get(PlatformSetting, _RAW_PURGE_WATERMARK_KEY) start_day: date | None = None if watermark_row is not None and isinstance(watermark_row.value, str): try: start_day = date.fromisoformat(watermark_row.value.strip()) + timedelta(days=1) except ValueError: start_day = None if start_day is None: start_day = await session.scalar( select(func.min(CmdbCostFact.date)), ) if start_day is None: # Empty table — nothing to purge, and no watermark to advance. return { "status": "ok", "days_processed": 0, "rows_purged": 0, "cutoff": cutoff.isoformat(), "retention_days": retention_days, } batch_size = max(1, min(int(batch_size), 10000)) max_days_per_run = max(1, min(int(max_days_per_run), 400)) max_seconds_per_run = max(30.0, min(float(max_seconds_per_run), 3600.0)) rows_purged = 0 days_processed = 0 day = start_day last_completed: date | None = None stopped_early: str | None = None started_monotonic = time.monotonic() while day < cutoff and days_processed < max_days_per_run: # Budget is checked BETWEEN days, never mid-day: a day is the # unit the watermark can express, so stopping inside one would # either lose the partial work or record a day as done when it # is not. Days are ~19k rows / ~3s here, so the overshoot past # the budget is bounded by one day's work. if time.monotonic() - started_monotonic >= max_seconds_per_run: stopped_early = "time_budget" break while True: # Two-step (SELECT ids → UPDATE ... WHERE id IN (...)) rather # than ``UPDATE ... WHERE id IN (SELECT ... LIMIT n)``: the # MySQL wire dialect rejects LIMIT inside an IN-subquery, and # the same two-step is what ``resource_directory.refresh_all`` # already uses for its history prune. ids = list((await session.scalars( select(CmdbCostFact.id) .where( CmdbCostFact.date == day, CmdbCostFact.raw.is_not(None), ) .order_by(CmdbCostFact.id) .limit(batch_size), )).all()) if not ids: break result = await session.execute( update(CmdbCostFact) .where(CmdbCostFact.id.in_(ids)) # ``sa_null()``, never ``None``. The model now also # carries ``none_as_null=True``, but this site states it # outright because getting it wrong here does not lose a # value — it loops forever: a JSON ``null`` still # satisfies this loop's own ``raw IS NOT NULL`` # predicate, so the same batch is re-selected until # something kills the tick (prod, 2026-09-02: 1800s, zero # rows freed). A one-line dependency on a type flag three # files away is not the right amount of coupling for a # statement whose failure mode is an infinite loop. .values(raw=sa_null()) .execution_options(synchronize_session=False), ) affected = int(getattr(result, "rowcount", 0) or 0) # Commit per batch: bounds the membuffer AND the transaction # duration (GC safepoint). See docstring. await session.commit() rows_purged += affected if len(ids) < batch_size: break last_completed = day days_processed += 1 day = day + timedelta(days=1) # Commit the watermark per DAY, not once at the end. A tick can # be killed at any instant (scheduler hard timeout, deploy, # SIGTERM); persisting only at the end threw away the whole run's # progress and — because ``days_processed`` counts empty days # too — left the next run re-walking the same prefix forever. await _write_raw_purge_watermark(session, last_completed) if days_processed >= max_days_per_run and day < cutoff: stopped_early = stopped_early or "day_budget" return { "status": "ok", "days_processed": days_processed, "rows_purged": rows_purged, "cutoff": cutoff.isoformat(), "retention_days": retention_days, "watermark": last_completed.isoformat() if last_completed else None, # Anchored on the actual position, not on "did we spend the day # budget" — the previous form reported ``caught_up`` for a run # that stopped on its budget one day short of the cutoff, and # will now also see time-budget stops. "caught_up": day >= cutoff, "stopped_early": stopped_early, } async def _write_raw_purge_watermark( session: AsyncSession, through: date, ) -> None: """Commit "raw is purged through ``through`` (inclusive)".""" await session.execute( mysql_insert(PlatformSetting) .values( key=_RAW_PURGE_WATERMARK_KEY, value=through.isoformat(), description=( "cmdb_cost_facts.raw 已清理到此日期(含)。由 " "cmdb.cost_facts_raw_purge 维护，勿手工编辑。" ), ) .on_duplicate_key_update(value=through.isoformat()), ) await session.commit() __all__ = [ "GROUP_BY_COLUMNS", "breakdown", "build_scope_lookup", "compute_fact_key", "create_allocation_rule", "delete_allocation_rule", "detect_anomalies", "export_facts_csv", "forecast_month_end", "ingest_bedrock_profile_facts", "ingest_cost_facts", "list_allocation_rules", "list_tag_keys", "resolve_raw_retention_days", "purge_cost_fact_raw", "recompute_allocation", "subordinates_cost_breakdown", "sync_bedrock_profile_facts_for_integration", "sync_cost_facts_for_integration", "timeseries", "top_resources", "update_allocation_rule", ]