"""Cloud server efficiency / cost-optimization engine. Turns the utilization rollups in ``cmdb_server_metrics_daily`` plus the per-resource spend in ``cmdb_cost_facts`` into one advisory efficiency profile per server (``cmdb_server_efficiency``): a rolling-window aggregate, a deterministic rule verdict, a recommendation, and an estimated monthly saving. Design invariants ----------------- * **Read-only.** Nothing here writes to any cloud. Verdicts and recommendations are advisory; execution stays in the existing ServerLifecycleActions surface. * **CPU + network are the primary signal.** They land on every cloud agent-free. Memory / disk usage need an in-guest agent and are often absent — so they only ever *strengthen* or *veto* a CPU-driven verdict, never gate it. A missing datapoint (``None``) is treated as "unknown", never as "0% / idle". * **Deterministic + explainable.** ``verdict_reasons`` records exactly which rule codes fired. AI narrative (a later PR) sits on top; the rule verdict must stand on its own. * **Configurable.** Thresholds default in code and are overridable via the ``cloud_efficiency.rules`` platform setting without a deploy. """ from __future__ import annotations import logging import math from collections.abc import Sequence from dataclasses import asdict, dataclass, field from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import ( EFFICIENCY_RECO_DOWNSIZE, EFFICIENCY_RECO_INVESTIGATE, EFFICIENCY_RECO_KEEP, EFFICIENCY_RECO_NONE, EFFICIENCY_RECO_STOP, EFFICIENCY_VERDICT_BUSY, EFFICIENCY_VERDICT_HEALTHY, EFFICIENCY_VERDICT_IDLE, EFFICIENCY_VERDICT_INSUFFICIENT_DATA, EFFICIENCY_VERDICT_NO_METRICS, EFFICIENCY_VERDICT_OVERSIZED, CmdbCostFact, CmdbServer, CmdbServerEfficiency, CmdbServerMetricsDaily, ) from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) RULES_SETTING_KEY = "cloud_efficiency.rules" # Cost is attributed from the trailing N days of line-item facts and # normalized to a 30-day month. COST_LOOKBACK_DAYS = 30 # Normalized statuses that mean the instance is NOT currently running, so a # stop/downsize recommendation would be a no-op / phantom saving. Kept as a # known-stopped allowlist (rather than "!= running") to avoid neutralizing a # genuinely-running box that reports a non-"running" transient/fallback state. NON_RUNNING_SERVER_STATUSES = frozenset({ "stopped", "stopping", "terminated", "terminating", "deallocated", "deallocating", "shutting-down", "shutdown", }) # ── Rule thresholds ────────────────────────────────────────────────── @dataclass class EfficiencyRules: """Tunable thresholds for the verdict engine. Defaults are conservative — they flag only clearly-wasteful boxes so the first recommendations a human sees are trustworthy. Overridable per-tenant via the ``cloud_efficiency.rules`` platform setting (any subset of keys).""" # Window + sampling window_days: int = 14 min_data_days: int = 3 # Idle (→ stop): peak-day CPU essentially flat AND network near-silent. idle_cpu_p95_max: float = 5.0 idle_net_bytes_max: float = 1_000_000.0 # avg in+out bytes / hour # Oversized (→ downsize): chronically underused but not fully idle. oversized_cpu_p95_max: float = 20.0 oversized_cpu_max_max: float = 40.0 # Busy (→ keep, possibly undersized): hot enough to leave alone. busy_cpu_p95_min: float = 80.0 # Memory veto/strengthen (only applied when memory is actually known): # * mem_p95 >= mem_high_pct → a low-CPU box is memory-bound, NOT # wasteful → never recommend stop/downsize. # * mem_p95 <= mem_idle_pct → strengthens an idle verdict. mem_high_pct: float = 50.0 mem_idle_pct: float = 20.0 # Savings model: a one-tier downsize conservatively saves ~half. downsize_savings_ratio: float = 0.5 rule_version: str = "v1" def merged(self, overrides: dict[str, Any] | None) -> EfficiencyRules: if not overrides: return self data = asdict(self) for k, v in overrides.items(): if k in data and v is not None: data[k] = v return EfficiencyRules(**data) async def load_rules(db: AsyncSession) -> EfficiencyRules: """Load rule thresholds: code defaults overlaid with the ``cloud_efficiency.rules`` platform setting when present.""" row = await db.get(PlatformSetting, RULES_SETTING_KEY) overrides = row.value if (row and isinstance(row.value, dict)) else None return EfficiencyRules().merged(overrides) # ── Aggregation ────────────────────────────────────────────────────── @dataclass class WindowStats: has_any_rows: bool = False data_days: int = 0 cpu_avg: float | None = None cpu_p95: float | None = None cpu_max: float | None = None mem_avg: float | None = None mem_p95: float | None = None mem_max: float | None = None net_in_avg: int | None = None net_out_avg: int | None = None def _mean(values: Sequence[float]) -> float | None: xs = [v for v in values if v is not None] if not xs: return None return sum(xs) / len(xs) def _percentile(values: list[float], q: float) -> float | None: """Linear-interpolation percentile (q in [0,1]) of the non-null values. Small-n friendly.""" xs = sorted(v for v in values if v is not None) if not xs: return None if len(xs) == 1: return xs[0] rank = q * (len(xs) - 1) lo = math.floor(rank) hi = math.ceil(rank) if lo == hi: return xs[int(lo)] frac = rank - lo return xs[int(lo)] * (1 - frac) + xs[int(hi)] * frac def aggregate_window(rows: list[CmdbServerMetricsDaily]) -> WindowStats: """Collapse a server's per-day rollups over the window into one stat bundle. ``data_days`` counts days that carried a real CPU sample — that's what the ``insufficient_data`` guard keys off.""" if not rows: return WindowStats(has_any_rows=False) cpu_days_avg = [r.cpu_pct_avg for r in rows if r.cpu_pct_avg is not None] cpu_days_p95 = [r.cpu_pct_p95 for r in rows if r.cpu_pct_p95 is not None] cpu_days_max = [r.cpu_pct_max for r in rows if r.cpu_pct_max is not None] mem_days_avg = [r.mem_pct_avg for r in rows if r.mem_pct_avg is not None] mem_days_p95 = [r.mem_pct_p95 for r in rows if r.mem_pct_p95 is not None] mem_days_max = [r.mem_pct_max for r in rows if r.mem_pct_max is not None] net_in = [r.net_in_bytes_avg for r in rows if r.net_in_bytes_avg is not None] net_out = [r.net_out_bytes_avg for r in rows if r.net_out_bytes_avg is not None] # A day counts toward data_days if it had at least one hourly sample # AND a CPU reading (CPU is the primary signal). data_days = sum( 1 for r in rows if (r.sample_count or 0) > 0 and r.cpu_pct_avg is not None ) net_in_mean = _mean(net_in) net_out_mean = _mean(net_out) return WindowStats( has_any_rows=True, data_days=data_days, cpu_avg=_mean(cpu_days_avg), # window p95 = typical *daily peak* (p95 across the daily p95s). cpu_p95=_percentile(cpu_days_p95, 0.95), cpu_max=max(cpu_days_max) if cpu_days_max else None, mem_avg=_mean(mem_days_avg), mem_p95=_percentile(mem_days_p95, 0.95), mem_max=max(mem_days_max) if mem_days_max else None, net_in_avg=int(net_in_mean) if net_in_mean is not None else None, net_out_avg=int(net_out_mean) if net_out_mean is not None else None, ) # ── Verdict ────────────────────────────────────────────────────────── @dataclass class Verdict: verdict: str reasons: list[str] = field(default_factory=list) recommendation: str = EFFICIENCY_RECO_NONE action_detail: dict[str, Any] | None = None def classify(stats: WindowStats, rules: EfficiencyRules) -> Verdict: """Deterministic verdict from the window stats. CPU + network drive it; memory only strengthens or vetoes.""" if not stats.has_any_rows: return Verdict(EFFICIENCY_VERDICT_NO_METRICS) if stats.data_days < rules.min_data_days or stats.cpu_p95 is None: return Verdict(EFFICIENCY_VERDICT_INSUFFICIENT_DATA) cpu_p95 = stats.cpu_p95 cpu_max = stats.cpu_max if stats.cpu_max is not None else cpu_p95 net_total = (stats.net_in_avg or 0) + (stats.net_out_avg or 0) mem_p95 = stats.mem_p95 mem_high = mem_p95 is not None and mem_p95 >= rules.mem_high_pct # 1) Idle candidate — peak-day CPU flat AND network near-silent. if cpu_p95 <= rules.idle_cpu_p95_max and net_total <= rules.idle_net_bytes_max: if mem_high: # Memory-bound box that happens to be CPU-light — do NOT stop. return Verdict( EFFICIENCY_VERDICT_HEALTHY, reasons=["cpu_p95_idle", "mem_bound"], recommendation=EFFICIENCY_RECO_KEEP, action_detail={"note": "mem_bound_despite_low_cpu"}, ) reasons = ["cpu_p95_idle", "net_idle"] if mem_p95 is not None and mem_p95 <= rules.mem_idle_pct: reasons.append("mem_low") return Verdict( EFFICIENCY_VERDICT_IDLE, reasons=reasons, recommendation=EFFICIENCY_RECO_STOP, action_detail={ "basis": "idle", "cpu_p95": round(cpu_p95, 2), "net_bytes_per_hour": net_total, }, ) # 2) Oversized candidate — chronically underused but not fully idle. if cpu_p95 <= rules.oversized_cpu_p95_max and cpu_max <= rules.oversized_cpu_max_max: if mem_high: return Verdict( EFFICIENCY_VERDICT_HEALTHY, reasons=["cpu_underused", "mem_bound"], recommendation=EFFICIENCY_RECO_INVESTIGATE, action_detail={"note": "mem_bound_review_before_downsize"}, ) return Verdict( EFFICIENCY_VERDICT_OVERSIZED, reasons=["cpu_underused"], recommendation=EFFICIENCY_RECO_DOWNSIZE, action_detail={ "basis": "cpu_p95", "cpu_p95": round(cpu_p95, 2), "cpu_max": round(cpu_max, 2), "savings_ratio": rules.downsize_savings_ratio, }, ) # 3) Busy — hot enough to leave alone (possibly undersized). if cpu_p95 >= rules.busy_cpu_p95_min: return Verdict( EFFICIENCY_VERDICT_BUSY, reasons=["cpu_busy"], recommendation=EFFICIENCY_RECO_KEEP, ) # 4) Healthy default. return Verdict( EFFICIENCY_VERDICT_HEALTHY, reasons=["cpu_normal"], recommendation=EFFICIENCY_RECO_KEEP, ) def estimate_savings( verdict: Verdict, monthly_cost: Decimal | None, rules: EfficiencyRules, ) -> Decimal | None: """Estimated monthly saving if the recommendation is acted on. ``None`` when cost couldn't be attributed to the instance.""" if monthly_cost is None: return None if verdict.recommendation == EFFICIENCY_RECO_STOP: return monthly_cost if verdict.recommendation == EFFICIENCY_RECO_DOWNSIZE: return (monthly_cost * Decimal(str(rules.downsize_savings_ratio))).quantize( Decimal("0.01") ) return Decimal("0.00") # ── Cost attribution ───────────────────────────────────────────────── @dataclass(frozen=True) class ResourceCost: """Per-resource attributed spend over the trailing window. ``usd`` is the converted total; ``currency`` is the resource's real billing currency (so a CNY-billed box that WAS converted is labeled ``CNY`` while ``monthly_cost_usd`` stays the USD figure), instead of stamping every attributed row ``USD`` unconditionally.""" usd: Decimal currency: str async def load_cost_map( db: AsyncSession, integration_id: str, *, today: date, lookback_days: int = COST_LOOKBACK_DAYS, ) -> dict[str, ResourceCost]: """One aggregation query → ``{resource_id: ResourceCost}`` over the trailing window for this integration. Instances are matched against these keys in-memory (exact, then ARN/path suffix) instead of running a leading-wildcard LIKE per server. 未换算 guard (mirrors the cost dashboard, see ``cost_service`` round-22 audit): a row whose ``currency != 'USD'`` but ``amount_usd == 0`` while ``amount_native != 0`` is an *unconverted native* line (the ingest driver couldn't convert — missing FX window). ``SUM(coalesce(amount_usd, 0))`` would silently fold it into a $0 total, turning a resource with real (foreign-currency) spend into an attributed ``$0`` instead of leaving it UNATTRIBUTED. A resource whose only spend is unconverted is therefore omitted from the map (→ ``monthly_cost`` stays NULL, savings unquantified) rather than mislabeled ``$0 USD``.""" start = today - timedelta(days=lookback_days) # Group by (resource_id, currency) so we can tell converted USD spend # apart from unconverted-native slices per resource. # 2026-07 成本口径改造:每实例月成本/可省切研发口径(业务折前 # 成本)。COALESCE(list, invoice) 过渡语义见 cost_view;回填后 # AWS SP 覆盖机器不再因抵扣行显示 ≈$0 月成本(「可省 $0」假象)。 from app.domains.cmdb.cost_view import fact_engineering_amount stmt = ( select( CmdbCostFact.resource_id, CmdbCostFact.currency, func.coalesce(func.sum(fact_engineering_amount()), 0), func.coalesce(func.sum(CmdbCostFact.amount_native), 0), ) .where( CmdbCostFact.integration_id == integration_id, CmdbCostFact.date >= start, CmdbCostFact.date <= today, CmdbCostFact.resource_id.is_not(None), ) .group_by(CmdbCostFact.resource_id, CmdbCostFact.currency) ) rows = (await db.execute(stmt)).all() # Fold per resource: total converted USD, whether it carries any # unconverted-native spend, and the currency bearing the most USD (its # real billing currency). agg: dict[str, dict[str, Any]] = {} for rid, currency, usd_sum, native_sum in rows: if not rid: continue cur = currency or "USD" usd = Decimal(str(usd_sum or 0)) native = Decimal(str(native_sum or 0)) a = agg.setdefault( rid, {"usd": Decimal("0"), "by_ccy": {}, "unconverted": False}, ) a["usd"] += usd if cur != "USD" and usd == 0 and native != 0: a["unconverted"] = True else: a["by_ccy"][cur] = a["by_ccy"].get(cur, Decimal("0")) + usd out: dict[str, ResourceCost] = {} for rid, a in agg.items(): if a["usd"] <= 0 and a["unconverted"]: # Only spend is unconverted native → leave UNATTRIBUTED. continue by_ccy: dict[str, Decimal] = a["by_ccy"] currency = ( max(by_ccy.items(), key=lambda kv: kv[1])[0] if by_ccy else "USD" ) out[rid] = ResourceCost(usd=a["usd"], currency=currency) return out def match_cost( candidates: str | list[str | None], cost_map: dict[str, Any], *, name_candidates: str | list[str | None] | None = None, ) -> Any | None: """Attribute a trailing-window spend to an instance. Billing line-item ``resource_id`` shapes vary by cloud (bare id, ARN, ARM resource path), and the identifier that appears in the cost export is NOT always the same string Helix stored as ``cmdb_servers.instance_id`` — most notably Azure, where ``instance_id`` is the ``vmId`` GUID but the cost ``resource_id`` is the ARM path ending in the VM *name*. ``candidates`` are IDENTITY keys (``instance_id``, the raw ARM id) — safe for exact matching. ``name_candidates`` are display names, which are user-controlled and NOT globally unique: a server named ``web`` could exact-collide with a DIFFERENT resource whose real cost ``resource_id`` is literally ``web``, stealing its spend. Names are therefore used ONLY in the ``/``- or ``:``-anchored, UNAMBIGUOUS suffix phase (the legitimate Azure "ARM path ends in the VM name" case), never exact. Returns ``None`` when nothing maps (utilization still renders; savings just can't be quantified).""" raw_ids: list[str | None] = ( [candidates] if isinstance(candidates, str) else list(candidates or []) ) ids: list[str] = [key for key in raw_ids if key] raw_names: list[str | None] = ( [name_candidates] if isinstance(name_candidates, str) else list(name_candidates or []) ) names: list[str] = [key for key in raw_names if key] if not ids and not names: return None # Exact / case-insensitive (Azure ARM ids are commonly lowercased in # CUR) — IDENTITY keys only. lower_map = {k.lower(): v for k, v in cost_map.items()} for key in ids: exact = cost_map.get(key) if exact is not None: return exact ci = lower_map.get(key.lower()) if ci is not None: return ci # Suffix phase — identity keys first, then names. for key in [*ids, *names]: if len(key) < 6: # guard: short tokens would match many paths continue kl = key.lower() matches = [ v for k, v in cost_map.items() if k.lower().endswith("/" + kl) or k.lower().endswith(":" + kl) ] # Only accept an UNAMBIGUOUS suffix match. Summing multiple matches # over-attributes cost: two distinct billed resources that share a # display-name suffix (e.g. two VMs both named ``web``) would have # their combined spend charged to one server, inflating its # monthly_cost and the idle→stop savings estimate — and the same # cost row could be claimed by two different servers, double-counting # the fleet total. If the suffix is ambiguous, treat as unattributed. if len(matches) == 1: return matches[0] return None def _normalize_month(cost: Decimal | None, lookback_days: int) -> Decimal | None: """Normalize a trailing-window total to a 30-day month.""" if cost is None: return None if lookback_days == 30: return cost.quantize(Decimal("0.01")) return (cost * Decimal(30) / Decimal(lookback_days)).quantize(Decimal("0.01")) # ── Orchestration ──────────────────────────────────────────────────── async def compute_efficiency_for_integration( db: AsyncSession, integration_id: str, *, today: date | None = None, narrate: bool = True, ) -> dict[str, Any]: """Recompute ``cmdb_server_efficiency`` for every live server of one integration. Idempotent upsert keyed on ``server_id``. Returns a summary dict for the scheduled-job result. When ``narrate`` is set (default) a best-effort AI-narrative pass runs on the top actionable instances after the deterministic compute — it no-ops when the LLM isn't configured and never raises (the rule verdict stands on its own).""" today = today or datetime.now(UTC).date() rules = await load_rules(db) # Inclusive window of exactly ``window_days`` dates ending today # (``>= start AND <= today``), so days=window_days-1. window_start = today - timedelta(days=max(0, rules.window_days - 1)) servers = ( await db.execute( select(CmdbServer).where( CmdbServer.integration_id == integration_id, CmdbServer.deleted_at.is_(None), ) ) ).scalars().all() if not servers: return {"status": "ok", "servers": 0, "computed": 0} server_ids = [str(s.id) for s in servers] # All window metric rows for these servers, grouped by server_id. metric_rows = ( await db.execute( select(CmdbServerMetricsDaily).where( CmdbServerMetricsDaily.integration_id == integration_id, CmdbServerMetricsDaily.metric_date >= window_start, CmdbServerMetricsDaily.metric_date <= today, ) ) ).scalars().all() by_server: dict[str, list[CmdbServerMetricsDaily]] = {} for r in metric_rows: by_server.setdefault(str(r.server_id), []).append(r) cost_map = await load_cost_map(db, integration_id, today=today) existing = ( await db.execute( select(CmdbServerEfficiency).where( CmdbServerEfficiency.server_id.in_(server_ids) ) ) ).scalars().all() existing_by_server = {str(e.server_id): e for e in existing} counts: dict[str, int] = {} total_savings = Decimal("0") computed = 0 now = datetime.now(UTC) for server in servers: sid = str(server.id) stats = aggregate_window(by_server.get(sid, [])) verdict = classify(stats, rules) # An already-stopped/terminated box may still be judged idle from the # running days earlier in the window — but recommending "stop" is a # no-op and "downsize" is moot, and counting full-cost savings on it # inflates the headline. Neutralize the action (keeps the verdict for # visibility; savings fall to 0 via recommendation=none). Key on a # KNOWN-stopped status set (not "!= running") so an Azure VM that # momentarily reports provisioningState 'succeeded' when powerState is # unavailable isn't wrongly neutralized while genuinely running. if ( (server.status or "").lower() in NON_RUNNING_SERVER_STATUSES and verdict.recommendation in ( EFFICIENCY_RECO_STOP, EFFICIENCY_RECO_DOWNSIZE, ) ): verdict.reasons = [*verdict.reasons, "instance_not_running"] verdict.recommendation = EFFICIENCY_RECO_NONE verdict.action_detail = { **(verdict.action_detail or {}), "note": "instance_not_running", } # Cost identifiers differ from instance_id on some clouds (esp. Azure: # instance_id=vmId GUID, cost resource_id=ARM path ending in name). # Identity keys (instance_id, raw ARM id) are exact-matchable; the # display name is used only in the unambiguous suffix phase (it is # user-controlled and could otherwise exact-collide with another # resource's real cost resource_id — see match_cost). raw = server.raw_data if isinstance(server.raw_data, dict) else {} cost_keys = [server.instance_id, raw.get("id")] matched = match_cost(cost_keys, cost_map, name_candidates=[server.name]) raw_cost = matched.usd if matched is not None else None monthly_cost = _normalize_month(raw_cost, COST_LOOKBACK_DAYS) savings = estimate_savings(verdict, monthly_cost, rules) if savings is not None: total_savings += savings counts[verdict.verdict] = counts.get(verdict.verdict, 0) + 1 computed += 1 row = existing_by_server.get(sid) if row is None: row = CmdbServerEfficiency(server_id=sid, integration_id=integration_id) db.add(row) row.integration_id = integration_id row.window_days = rules.window_days row.data_days = stats.data_days row.cpu_avg = stats.cpu_avg row.cpu_p95 = stats.cpu_p95 row.cpu_max = stats.cpu_max row.mem_avg = stats.mem_avg row.mem_p95 = stats.mem_p95 row.mem_max = stats.mem_max row.net_in_avg = stats.net_in_avg row.net_out_avg = stats.net_out_avg row.verdict = verdict.verdict row.verdict_reasons = verdict.reasons row.recommendation = verdict.recommendation row.recommended_action_detail = verdict.action_detail row.monthly_cost_usd = monthly_cost row.estimated_monthly_savings_usd = savings # Stamp the resource's real billing currency (not an unconditional # "USD"). NULL when unattributed — including an unconverted-native # resource the 未换算 guard in load_cost_map dropped from the map. row.cost_currency = ( matched.currency if matched is not None and monthly_cost is not None else None ) # Spec snapshot for the console-savings detector (migration 425) — # tomorrow's run compares the freshly-synced spec against this to # recognise an out-of-band downsize of a flagged box. row.instance_type = server.instance_type row.vcpu = server.vcpu row.memory_mb = server.memory_mb row.rule_version = rules.rule_version row.computed_at = now await db.commit() narrated: dict[str, Any] | None = None if narrate: try: from app.domains.cmdb.efficiency_ai import generate_narratives narrated = await generate_narratives(db, integration_id) except Exception as exc: # noqa: BLE001 — AI is strictly best-effort logger.warning( "cloud-efficiency narration pass failed for %s: %s", integration_id, str(exc)[:200], ) await db.rollback() return { "status": "ok", "servers": len(servers), "computed": computed, "verdict_counts": counts, "estimated_monthly_savings_usd": float(total_savings), "narrated": narrated, }