"""Task: Worker fleet health check + auto-scaling loop (runs every 60s via arq cron). Per the locked design: 1. CF Queue consumers: read depth → adjust max_concurrency via CF API if above/below thresholds 2. CF Workers: ping health URL → update health status, alert on failure 3. Pending deploys: workers where target_version != current_version → fire CI webhook """ from __future__ import annotations import structlog from datetime import datetime, timezone import httpx logger = structlog.get_logger() async def monitor_worker_fleet(ctx: dict) -> dict: from src.core.database import async_session_factory from src.domain.fleet.repository import ( WorkerFleetEventRepository, WorkerFleetRegistryRepository, WorkerScalingConfigRepository, ) from src.domain.fleet.service import ( FleetService, get_cf_queue_depth, _set_cf_consumer_concurrency, ) db_factory = ctx.get("db_factory", async_session_factory) scaled_up = scaled_down = health_failed = health_recovered = deploys_triggered = 0 async with db_factory() as db: registry = WorkerFleetRegistryRepository(db) events = WorkerFleetEventRepository(db) cfg_repo = WorkerScalingConfigRepository(db) cfg = await cfg_repo.get() if not cfg or not cfg.auto_scale_enabled: logger.debug("fleet_monitor_skip", reason="auto_scale_disabled") return {"status": "skipped"} now = datetime.now(timezone.utc) # ---------------------------------------------------------------- # 1. Queue consumers — depth-based concurrency scaling # ---------------------------------------------------------------- consumers = await registry.list_queue_consumers() for worker in consumers: if not worker.cf_queue_id: continue try: depth = await get_cf_queue_depth(worker.cf_queue_id) except Exception as exc: logger.warning( "fleet_queue_depth_error", worker=worker.worker_name, error=str(exc), ) continue current = worker.current_concurrency or worker.min_concurrency or 1 max_c = worker.max_concurrency or 20 min_c = worker.min_concurrency or 1 up_thresh = worker.scale_up_threshold or 100 down_thresh = worker.scale_down_threshold or 10 if depth > up_thresh and current < max_c: new_concurrency = min(current * 2, max_c) try: await _set_cf_consumer_concurrency(worker, new_concurrency) await registry.update( worker, current_concurrency=new_concurrency, updated_at=now ) await events.create( worker_name=worker.worker_name, instance_id=worker.instance_id, event_type="scaled_up", triggered_by="auto_scaler", from_value=str(current), to_value=str(new_concurrency), metadata={"queue_depth": depth}, ) scaled_up += 1 logger.info( "fleet_scaled_up", worker=worker.worker_name, from_c=current, to_c=new_concurrency, depth=depth, ) except Exception as exc: logger.error( "fleet_scale_up_error", worker=worker.worker_name, error=str(exc), ) elif depth < down_thresh and current > min_c: new_concurrency = max(current // 2, min_c) try: await _set_cf_consumer_concurrency(worker, new_concurrency) await registry.update( worker, current_concurrency=new_concurrency, updated_at=now ) await events.create( worker_name=worker.worker_name, instance_id=worker.instance_id, event_type="scaled_down", triggered_by="auto_scaler", from_value=str(current), to_value=str(new_concurrency), metadata={"queue_depth": depth}, ) scaled_down += 1 logger.info( "fleet_scaled_down", worker=worker.worker_name, from_c=current, to_c=new_concurrency, depth=depth, ) except Exception as exc: logger.error( "fleet_scale_down_error", worker=worker.worker_name, error=str(exc), ) # ---------------------------------------------------------------- # 2. CF Workers — health ping # ---------------------------------------------------------------- cf_workers = await registry.list_cf_workers() for worker in cf_workers: if not worker.health_url: continue was_ok = worker.last_health_check_ok try: async with httpx.AsyncClient(timeout=5) as client: resp = await client.get(worker.health_url) is_ok = resp.status_code < 500 except Exception: is_ok = False await registry.update( worker, last_health_check_at=now, last_health_check_ok=is_ok, status="active" if is_ok else "unhealthy", updated_at=now, ) if not is_ok and was_ok is not False: await events.create( worker_name=worker.worker_name, instance_id=worker.instance_id, event_type="health_failed", triggered_by="auto_scaler", ) health_failed += 1 logger.warning( "fleet_health_failed", worker=worker.worker_name, instance=worker.instance_id, url=worker.health_url, ) elif is_ok and was_ok is False: await events.create( worker_name=worker.worker_name, instance_id=worker.instance_id, event_type="recovered", triggered_by="auto_scaler", ) health_recovered += 1 logger.info( "fleet_health_recovered", worker=worker.worker_name, instance=worker.instance_id, ) # ---------------------------------------------------------------- # 3. Pending version deployments # ---------------------------------------------------------------- if cfg.deployment_webhook_url: pending = await registry.list_pending_deploy() for worker in pending: try: svc = FleetService( registry=registry, events=events, config=cfg_repo, ) await svc._fire_deployment_webhook( cfg, worker.worker_name, worker.instance_id, worker.target_version, ) await registry.update(worker, status="deploying", updated_at=now) await events.create( worker_name=worker.worker_name, instance_id=worker.instance_id, event_type="deploy_triggered", triggered_by="auto_scaler", from_value=worker.current_version, to_value=worker.target_version, ) deploys_triggered += 1 logger.info( "fleet_auto_deploy_triggered", worker=worker.worker_name, version=worker.target_version, ) except Exception as exc: logger.error( "fleet_auto_deploy_error", worker=worker.worker_name, error=str(exc), ) await db.commit() return { "status": "ok", "scaled_up": scaled_up, "scaled_down": scaled_down, "health_failed": health_failed, "health_recovered": health_recovered, "deploys_triggered": deploys_triggered, }