"""Add ``cmdb_cost_facts`` (line-item ground truth) + ``cmdb_cost_allocation_rules``. Production-grade FinOps upgrade (2026-05-09). The existing ``cmdb_billing_monthly`` and ``cmdb_cloud_billing_daily`` tables aggregate cost at the (integration × month) and (integration × scope × date) levels respectively — coarse enough that a finance lead asking "which Azure resource group inside subscription 7bc8ab17 spent what last month" has no answer except "open the Azure portal." Every cloud's billing API exposes line-item granularity (Azure Cost Management ``ResourceId`` grouping, Aliyun BSS ``DescribeInstanceBill``, AWS Cost Explorer ``GetCostAndUsageWithResources``, Tencent ``DescribeBillDetail``, GCP BigQuery export); the gap is on the Helix side where we throw it away. This migration adds the ground-truth fact table: * ``cmdb_cost_facts`` — one row per (date × resource × usage_type) with full multi-dimensional shape: resource_group, resource_id, resource_type, region, service, usage_type, billing-side tags, amounts in both USD and native currency, usage quantity, and denormalized cost-owner columns (person/department/cost_center) populated at ingest time from allocation rules so analytics queries don't have to re-evaluate rules per row. * ``cmdb_cost_allocation_rules`` — priority-ordered rule ladder for attributing cost facts to a person / department / cost-center based on integration, provider, resource_group, tag key/value, or service. The first matching rule wins; unmatched rows surface as "unallocated" in the dashboard, which is the FinOps-industry-standard KPI. Schema notes: * ``id BIGINT AUTO_INCREMENT`` on ``cmdb_cost_facts`` — at scale this table holds tens of millions of rows; UUID PKs cost ~24 bytes/row of index space and force every secondary-index lookup back through a random-order PK lookup. BIGINT (8 bytes) keeps the clustered index tight and TiDB's IndexLookup plans clean. * ``fact_key CHAR(64) UNIQUE`` — sha256 hex of the natural key (date|integration|scope|resource_id|usage_type). Since some components legitimately can be NULL (resource_id on org-level lines like Azure "Cost Anomaly Detection", usage_type when the provider doesn't break SKUs out), a hash collapses NULLs into a deterministic string and lets MySQL's ``INSERT ... ON DUPLICATE KEY UPDATE`` do idempotent re-ingestion. * ``amount_usd DECIMAL(14,6)`` — six-digit precision because GCP BigQuery export and Aliyun BSS both return micro-dollar amounts on per-second-billed resources. Truncating to two decimals would skew a "top 100 cheapest resources" listing. * All FK columns to ``integrations.id`` go through ``normalize_fk_column_collation`` because TiDB rejects CHAR↔VARCHAR pairings that local MySQL accepts (memory: ``feedback_tidb_fk_strictness``). * RESTRICT (not CASCADE) on the integration FK — same reasoning as migration 124 / 122: a hand-typed ``DELETE FROM integrations`` must not silently wipe years of historical cost facts. Operators soft-disable via ``Integration.enabled=False``. Indexes are sized to the dashboard's query shape: * ``ix_facts_date_int`` — primary trend / KPI rollup (every cost query starts from a date range + integration filter). * ``ix_facts_resource`` — resource detail drawer (top-N resources with sparkline). * ``ix_facts_int_rg_date`` — Azure RG / GCP folder / Aliyun member drilldown. * ``ix_facts_int_service_date`` — service-pivot dashboard. * ``ix_facts_owner_date`` — chargeback view by person. * ``ix_facts_cc_date`` — chargeback view by cost center. Revision ID: 139 Revises: 138 Create Date: 2026-05-09 """ from __future__ import annotations import hashlib import json as _json import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import normalize_fk_column_collation # revision identifiers, used by Alembic. revision = "139" down_revision = "138" branch_labels = None depends_on = None def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n" ), {"n": name}, ).scalar() is not None def _fk_exists(bind, table: str, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " "AND CONSTRAINT_NAME = :n AND CONSTRAINT_TYPE = 'FOREIGN KEY'" ), {"t": table, "n": name}, ).scalar() is not None def _index_exists(bind, table: str, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " "AND INDEX_NAME = :n" ), {"t": table, "n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() # ── cmdb_cost_facts ────────────────────────────────────────────── if not _table_exists(bind, "cmdb_cost_facts"): op.create_table( "cmdb_cost_facts", sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True), # 64-hex sha256 of the natural key — idempotent upsert handle. sa.Column("fact_key", sa.CHAR(64), nullable=False), sa.Column("date", sa.Date, nullable=False), sa.Column("integration_id", sa.String(36), nullable=False), sa.Column("scope_id", sa.String(36), nullable=True), # Azure RG / GCP folder / Aliyun member account / AWS OU pseudonym. sa.Column("resource_group", sa.String(256), nullable=True), # Full provider id: ARN / Azure resource id / GCP relative # name / Aliyun InstanceID. NULL for line items not tied to # a single resource (Azure "Cost Anomaly Detection", # tenant-wide overhead lines). sa.Column("resource_id", sa.String(512), nullable=True), sa.Column("resource_name", sa.String(256), nullable=True), sa.Column("resource_type", sa.String(128), nullable=True), sa.Column("region", sa.String(64), nullable=True), # MeterCategory / ServiceName — the user-facing service label # (always present; "Other" when provider doesn't classify). sa.Column("service", sa.String(128), nullable=False, server_default=""), sa.Column("usage_type", sa.String(256), nullable=True), sa.Column("tags", sa.JSON, nullable=True), sa.Column( "amount_usd", sa.Numeric(14, 6), nullable=False, server_default=sa.text("0"), ), sa.Column( "amount_native", sa.Numeric(14, 6), nullable=False, server_default=sa.text("0"), ), sa.Column("currency", sa.String(8), nullable=False, server_default="USD"), sa.Column("usage_quantity", sa.Numeric(20, 6), nullable=True), sa.Column("usage_unit", sa.String(64), nullable=True), # Allocation outputs — denormalized at ingest so analytics # queries don't re-walk the rule ladder. Recomputable via # ``POST /cmdb/cost/allocation-rules/recompute``. sa.Column("cost_owner_person_id", sa.String(36), nullable=True), sa.Column("cost_owner_department", sa.String(256), nullable=True), sa.Column("cost_center", sa.String(64), nullable=True), sa.Column("allocation_rule_id", sa.String(36), nullable=True), sa.Column("raw", sa.JSON, nullable=True), sa.Column( "fetched_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), onupdate=sa.func.now(), nullable=False, ), sa.UniqueConstraint("fact_key", name="uq_cmdb_cost_facts_fact_key"), mysql_engine="InnoDB", mysql_charset="utf8mb4", mysql_collate="utf8mb4_general_ci", ) # Normalise FK column collation BEFORE creating the FK so TiDB # accepts the cross-table join. normalize_fk_column_collation( table="cmdb_cost_facts", column="integration_id", parent_table="integrations", nullable=False, ) normalize_fk_column_collation( table="cmdb_cost_facts", column="scope_id", parent_table="cmdb_cloud_scopes", nullable=True, ) normalize_fk_column_collation( table="cmdb_cost_facts", column="cost_owner_person_id", parent_table="identity_persons", nullable=True, ) op.create_foreign_key( "fk_cmdb_cost_facts_integration", source_table="cmdb_cost_facts", referent_table="integrations", local_cols=["integration_id"], remote_cols=["id"], ondelete="RESTRICT", ) op.create_foreign_key( "fk_cmdb_cost_facts_scope", source_table="cmdb_cost_facts", referent_table="cmdb_cloud_scopes", local_cols=["scope_id"], remote_cols=["id"], ondelete="SET NULL", ) op.create_foreign_key( "fk_cmdb_cost_facts_owner", source_table="cmdb_cost_facts", referent_table="identity_persons", local_cols=["cost_owner_person_id"], remote_cols=["id"], ondelete="SET NULL", ) if not _index_exists(bind, "cmdb_cost_facts", "ix_facts_date_int"): op.create_index( "ix_facts_date_int", "cmdb_cost_facts", ["date", "integration_id"], ) if not _index_exists(bind, "cmdb_cost_facts", "ix_facts_resource"): op.create_index( "ix_facts_resource", "cmdb_cost_facts", ["resource_id", "date"], ) if not _index_exists(bind, "cmdb_cost_facts", "ix_facts_int_rg_date"): op.create_index( "ix_facts_int_rg_date", "cmdb_cost_facts", ["integration_id", "resource_group", "date"], ) if not _index_exists(bind, "cmdb_cost_facts", "ix_facts_int_service_date"): op.create_index( "ix_facts_int_service_date", "cmdb_cost_facts", ["integration_id", "service", "date"], ) if not _index_exists(bind, "cmdb_cost_facts", "ix_facts_owner_date"): op.create_index( "ix_facts_owner_date", "cmdb_cost_facts", ["cost_owner_person_id", "date"], ) if not _index_exists(bind, "cmdb_cost_facts", "ix_facts_cc_date"): op.create_index( "ix_facts_cc_date", "cmdb_cost_facts", ["cost_center", "date"], ) # ── cmdb_cost_allocation_rules ────────────────────────────────── if not _table_exists(bind, "cmdb_cost_allocation_rules"): op.create_table( "cmdb_cost_allocation_rules", sa.Column("id", sa.String(36), primary_key=True), sa.Column("name", sa.String(255), nullable=False), sa.Column( "priority", sa.Integer, nullable=False, server_default=sa.text("100"), ), # Match conditions — NULL means "match any". Multiple # non-NULL conditions are AND-ed. ``match_integration_ids`` # is a JSON array because a rule may apply to a set of # integrations (e.g. "all PingCAP Aliyun accounts" or "the # two Azure subs the data team uses"). sa.Column("match_integration_ids", sa.JSON, nullable=True), sa.Column("match_provider", sa.String(32), nullable=True), sa.Column("match_resource_group", sa.String(256), nullable=True), sa.Column("match_service", sa.String(128), nullable=True), sa.Column("match_tag_key", sa.String(128), nullable=True), sa.Column("match_tag_value", sa.String(256), nullable=True), # Assignment outputs — at least one must be non-NULL. sa.Column("assign_person_id", sa.String(36), nullable=True), sa.Column("assign_department", sa.String(256), nullable=True), sa.Column("assign_cost_center", sa.String(64), nullable=True), sa.Column( "enabled", sa.Boolean, nullable=False, server_default=sa.text("1"), ), sa.Column("description", sa.Text, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column("created_by_id", sa.String(36), nullable=True), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), onupdate=sa.func.now(), nullable=False, ), sa.Column("updated_by_id", sa.String(36), nullable=True), mysql_engine="InnoDB", mysql_charset="utf8mb4", mysql_collate="utf8mb4_general_ci", ) normalize_fk_column_collation( table="cmdb_cost_allocation_rules", column="assign_person_id", parent_table="identity_persons", nullable=True, ) op.create_foreign_key( "fk_cmdb_cost_alloc_assign_person", source_table="cmdb_cost_allocation_rules", referent_table="identity_persons", local_cols=["assign_person_id"], remote_cols=["id"], ondelete="SET NULL", ) if not _index_exists( bind, "cmdb_cost_allocation_rules", "ix_alloc_rules_priority", ): op.create_index( "ix_alloc_rules_priority", "cmdb_cost_allocation_rules", ["enabled", "priority"], ) # ── Seed per-(cloud-integration) cost_facts_sync jobs ──────────── # # Mirrors migration 052's seed of ``cmdb.billing_daily.{slug}`` — # one cron row per cloud integration with billing capability so # the admin "Scheduled Jobs" page lists them out of the box and # operators can disable / retune individual integrations without # touching the global ``cmdb.cost_facts_sync`` handler. if ( _table_exists(bind, "scheduled_jobs") and _table_exists(bind, "integrations") ): cloud_rows = bind.execute( sa.text( "SELECT id, name, slug, capabilities FROM integrations " "WHERE kind = 'cloud'" ), ).fetchall() for integ in cloud_rows: integ_id = str(integ[0]) integ_name = integ[1] slug = integ[2] or integ_id caps = integ[3] if isinstance(caps, str): try: caps_list = _json.loads(caps) except (ValueError, TypeError): caps_list = [] elif isinstance(caps, list): caps_list = caps else: caps_list = [] # Only seed for integrations that opt into billing. if "billing" not in caps_list: continue job_name = f"cmdb.cost_facts.{slug}" offset = int( hashlib.sha256(job_name.encode()).hexdigest(), 16, ) % 60 bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " "trigger_type, cron_expr, timezone, enabled, config, " "run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE name = :name)" ), { "id": str(uuid.uuid4()), "name": job_name, "display": f"{integ_name} · 成本明细抓取", "desc": ( "每日 04:00 拉取昨天的 line-item 成本明细并写入 " "cmdb_cost_facts。" ), "handler": "cmdb.cost_facts_sync", "cron": f"{offset} 4 * * *", "config": _json.dumps({"integration_id": integ_id}), }, ) def downgrade() -> None: bind = op.get_bind() # Reverse the cron seeding. Without this, a downgrade leaves # orphaned ``cmdb.cost_facts.{slug}`` rows in scheduled_jobs # that re-arm on the NEXT upgrade as if they were always there # — confusing and breaks the migration's idempotency contract. if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "DELETE FROM scheduled_jobs " "WHERE handler_key = 'cmdb.cost_facts_sync'" ), ) # Drop FKs BEFORE indexes — MySQL/TiDB refuse to drop an index # that backs a still-existing FK constraint. if _table_exists(bind, "cmdb_cost_allocation_rules"): if _fk_exists( bind, "cmdb_cost_allocation_rules", "fk_cmdb_cost_alloc_assign_person", ): op.drop_constraint( "fk_cmdb_cost_alloc_assign_person", "cmdb_cost_allocation_rules", type_="foreignkey", ) for idx in ("ix_alloc_rules_priority",): if _index_exists(bind, "cmdb_cost_allocation_rules", idx): op.drop_index(idx, table_name="cmdb_cost_allocation_rules") op.drop_table("cmdb_cost_allocation_rules") if _table_exists(bind, "cmdb_cost_facts"): for fk in ( "fk_cmdb_cost_facts_owner", "fk_cmdb_cost_facts_scope", "fk_cmdb_cost_facts_integration", ): if _fk_exists(bind, "cmdb_cost_facts", fk): op.drop_constraint( fk, "cmdb_cost_facts", type_="foreignkey", ) for idx in ( "ix_facts_cc_date", "ix_facts_owner_date", "ix_facts_int_service_date", "ix_facts_int_rg_date", "ix_facts_resource", "ix_facts_date_int", ): if _index_exists(bind, "cmdb_cost_facts", idx): op.drop_index(idx, table_name="cmdb_cost_facts") op.drop_table("cmdb_cost_facts")