"""Pull cloud instance utilization and roll it up into ``cmdb_server_metrics_daily``. Called by the ``cmdb.server_metrics_sync`` scheduled handler once per day per cloud integration that declares the ``metrics`` capability. Reads the already-synced ``cmdb_servers`` fleet (never re-lists it from the cloud), asks the provider's metrics adapter for the trailing window of hourly datapoints, buckets them by UTC date, and idempotently upserts one row per (server, day). Prunes rollups older than the retention horizon. **Three phases, and the connection is released between them.** The cloud leg is *minutes* long (2026-08-10 prod: 27 min for AliCloud 海外 — 583 instances × 7 CloudMonitor metrics × master + every RD member account), so it must not run inside a database transaction: 1. read the fleet, then ``commit()`` — hands the pooled connection back; 2. pull from the cloud holding **no** connection and no snapshot; 3. re-acquire a connection and upsert in bounded, committed chunks. Before this split, phase 1's read transaction stayed open across phase 2 and the first statement after the pull was the one that paid: the 2026-08-10 failure was errno 2013 ("Lost connection to MySQL server during query") on the preload ``SELECT cmdb_server_metrics_daily ...`` after a 24-minute idle gap, which threw away a completed 24-minute pull. ``pool_pre_ping`` cannot help there — it probes at *checkout*, and this connection was checked out 24 minutes earlier. It also pinned one of the 10 pool slots (and a >GC-life-time read snapshot) for the whole pull. """ from __future__ import annotations import logging import uuid from collections import defaultdict from collections.abc import Sequence from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import Row, delete, func, select from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.efficiency import _mean, _percentile from app.domains.cmdb.models import CmdbServer, CmdbServerMetricsDaily from app.domains.cmdb.provider_factory import build_cloud_provider from app.domains.cmdb.providers.base import ( ProviderCapability, ServerMetricPoint, ServerMetricTarget, ) from app.domains.cmdb.providers.exceptions import CapabilityNotSupported logger = logging.getLogger(__name__) DEFAULT_LOOKBACK_DAYS = 14 RETENTION_DAYS = 90 PERIOD_SECONDS = 3600 # Rows per upsert statement / transaction in phase 3. AliCloud 海外 # writes ~8 000 (server × day) rows per run; a single statement + single # commit for the lot makes one dropped connection cost the entire pull # again, and grows the TiDB transaction membuffer for no benefit. Chunked # commits make progress durable and keep every statement small. Safe to # chunk because each chunk is an independent idempotent upsert keyed on # ``uq_server_metrics_daily`` — a re-run overwrites, never duplicates. _UPSERT_CHUNK_ROWS = 500 # Columns the daily rollup owns. Written on INSERT and re-written on # duplicate-key so a same-day re-run corrects yesterday's partial pull. # ``sample_count`` included deliberately: a later run with MORE hourly # buckets must be able to lower/raise it, and the efficiency engine reads # it to detect thin sampling. _ROLLUP_COLUMNS = ( "cpu_pct_avg", "cpu_pct_p95", "cpu_pct_max", "mem_pct_avg", "mem_pct_p95", "mem_pct_max", "disk_used_pct_avg", "disk_used_pct_max", "net_in_bytes_avg", "net_out_bytes_avg", "disk_read_bytes_avg", "disk_write_bytes_avg", "sample_count", ) def _targets_from_servers( servers: Sequence[CmdbServer | Row[Any]], ) -> list[ServerMetricTarget]: """Build metric targets from fleet rows. Accepts ORM instances or the lightweight column ``Row`` tuples phase 1 selects — both expose the same attribute names. """ return [ ServerMetricTarget( instance_id=s.instance_id, region=s.region, zone=s.zone, project_id=s.project_id, raw=s.raw_data if isinstance(s.raw_data, dict) else None, ) for s in servers if s.instance_id ] def _rollup_points(points: list[ServerMetricPoint]) -> dict[tuple[str, Any], dict]: """Bucket hourly points into per-(instance_id, date) daily rollups.""" buckets: dict[tuple[str, Any], list[ServerMetricPoint]] = defaultdict(list) for p in points: buckets[(p.instance_id, p.timestamp.date())].append(p) rollups: dict[tuple[str, Any], dict] = {} for (instance_id, day), pts in buckets.items(): cpu = [p.cpu_pct for p in pts if p.cpu_pct is not None] cpu_max_src = [ (p.cpu_pct_max if p.cpu_pct_max is not None else p.cpu_pct) for p in pts if (p.cpu_pct_max is not None or p.cpu_pct is not None) ] mem = [p.mem_pct for p in pts if p.mem_pct is not None] disk = [p.disk_used_pct for p in pts if p.disk_used_pct is not None] net_in = [p.net_in_bytes for p in pts if p.net_in_bytes is not None] net_out = [p.net_out_bytes for p in pts if p.net_out_bytes is not None] d_read = [p.disk_read_bytes for p in pts if p.disk_read_bytes is not None] d_write = [p.disk_write_bytes for p in pts if p.disk_write_bytes is not None] net_in_mean = _mean(net_in) net_out_mean = _mean(net_out) d_read_mean = _mean(d_read) d_write_mean = _mean(d_write) rollups[(instance_id, day)] = { "cpu_pct_avg": _mean(cpu), "cpu_pct_p95": _percentile(cpu, 0.95), "cpu_pct_max": max(cpu_max_src) if cpu_max_src else None, "mem_pct_avg": _mean(mem), "mem_pct_p95": _percentile(mem, 0.95), "mem_pct_max": max(mem) if mem else None, "disk_used_pct_avg": _mean(disk), "disk_used_pct_max": max(disk) if disk else None, "net_in_bytes_avg": int(net_in_mean) if net_in_mean is not None else None, "net_out_bytes_avg": int(net_out_mean) if net_out_mean is not None else None, "disk_read_bytes_avg": int(d_read_mean) if d_read_mean is not None else None, "disk_write_bytes_avg": int(d_write_mean) if d_write_mean is not None else None, "sample_count": len(pts), } return rollups async def _flush_rollup_chunk( db: AsyncSession, chunk: list[dict[str, Any]], ) -> None: """Upsert one bounded chunk of daily rollups and commit it. ``INSERT ... ON DUPLICATE KEY UPDATE`` on ``uq_server_metrics_daily`` replaces the previous read-modify-write (preload every existing row in the window into the identity map, then ``setattr``). That preload was the statement the 2026-08-10 drop killed, and it grew with the fleet; the upsert needs no read at all. ``created_at`` is deliberately NOT in the SET list — first-seen must survive a re-run. """ stmt = mysql_insert(CmdbServerMetricsDaily).values(chunk) await db.execute( stmt.on_duplicate_key_update( **{c: stmt.inserted[c] for c in _ROLLUP_COLUMNS}, source=stmt.inserted.source, updated_at=func.now(), ) ) await db.commit() async def _write_rollups( db: AsyncSession, *, integration_id: str, rollups: dict[tuple[str, Any], dict], instance_to_server: dict[str, str], source: str, prune_before: Any, ) -> dict[str, int]: """Phase 3 — chunked idempotent upsert + retention prune. Owns its commits and re-derives everything from its arguments, so the caller can re-run it wholesale on a fresh session after a transient DB error (1205 / 1213 / 2006 / 2013) without re-doing the cloud pull. """ upserted = 0 skipped_unmapped = 0 chunk: list[dict[str, Any]] = [] for (instance_id, day), vals in rollups.items(): server_id = instance_to_server.get(instance_id) if server_id is None: # Metrics for an instance not in our live fleet (race with a # just-terminated box, or a filtered instance) — ignore. skipped_unmapped += 1 continue chunk.append({ "id": str(uuid.uuid4()), "server_id": server_id, "integration_id": integration_id, "metric_date": day, "source": source, **vals, }) if len(chunk) >= _UPSERT_CHUNK_ROWS: await _flush_rollup_chunk(db, chunk) upserted += len(chunk) chunk = [] if chunk: await _flush_rollup_chunk(db, chunk) upserted += len(chunk) # Retention prune (this integration only). prune_result = await db.execute( delete(CmdbServerMetricsDaily).where( CmdbServerMetricsDaily.integration_id == integration_id, CmdbServerMetricsDaily.metric_date < prune_before, ) ) await db.commit() return { "days_upserted": upserted, "unmapped_instances": skipped_unmapped, "pruned": prune_result.rowcount or 0, } async def run_metrics_sync( db: AsyncSession, integration_id: str, *, lookback_days: int = DEFAULT_LOOKBACK_DAYS, ) -> dict[str, Any]: """Pull + upsert daily utilization rollups for one integration. Returns a summary dict. Raises on hard failure (bad credentials, provider error) so the scheduled-job run lands ``status=failed``; returns ``{"status": "skipped"}`` for structural no-ops (provider lacks the metrics capability, no servers). """ # Lazy import: ``app.scheduled_jobs.handlers`` imports this module. from app.scheduled_jobs.handlers import run_with_lock_retry provider, alias = await build_cloud_provider(db, integration_id) try: if not provider.supports(ProviderCapability.METRICS): return {"status": "skipped", "reason": "metrics_not_supported"} # ── Phase 1: read the fleet ─────────────────────────────────── # Column select, not ORM entities: nothing here survives into # phase 3, and plain ``Row`` tuples can't be expired by the # commit below (an expired ORM attribute would re-fetch — i.e. # re-acquire a connection — from inside the cloud leg). servers = ( await db.execute( select( CmdbServer.id, CmdbServer.instance_id, CmdbServer.region, CmdbServer.zone, CmdbServer.project_id, CmdbServer.raw_data, ).where( CmdbServer.integration_id == integration_id, CmdbServer.deleted_at.is_(None), ) ) ).all() # Hand the pooled connection back BEFORE the multi-minute cloud # leg. Nothing was written, but the implicit read transaction is # still open, and an open transaction holds its connection. await db.commit() if not servers: return {"status": "skipped", "reason": "no_servers"} # instance_id → server_id (only live servers get metrics rows). instance_to_server = { s.instance_id: str(s.id) for s in servers if s.instance_id } targets = _targets_from_servers(servers) # ── Phase 2: cloud pull, no DB connection held ──────────────── try: points = await provider.list_server_metrics( targets, lookback_days=lookback_days, period_seconds=PERIOD_SECONDS, ) except CapabilityNotSupported: return {"status": "skipped", "reason": "metrics_not_implemented"} partial_failures = provider.consume_partial_failures() skipped_legs = provider.consume_skipped_legs() rollups = _rollup_points(points) # ── Phase 3: write ─────────────────────────────────────────── # Fresh connection (pre-pinged at checkout) + one bounded retry # family for a genuine mid-statement drop, so a TiDB node recycle # during the write costs a re-run of the write, not of the pull. today = datetime.now(UTC).date() written = await run_with_lock_retry( db, lambda session: _write_rollups( session, integration_id=integration_id, rollups=rollups, instance_to_server=instance_to_server, source=_source_for_provider(provider.provider_type()), prune_before=today - timedelta(days=RETENTION_DAYS), ), label=f"cmdb.server_metrics_sync write {alias}", ) return { "status": "ok", "integration": alias, "servers": len(servers), "points": len(points), "partial_failures": partial_failures, "skipped_legs": skipped_legs, **written, } finally: await provider.close() def _source_for_provider(provider_type: str) -> str: return { "aws": "cloudwatch", "aliyun": "aliyun_cms", "alicloud_intl": "aliyun_cms", "azure": "azure_monitor", "gcp": "gcp_monitoring", }.get(provider_type, provider_type)