Worker Monitoring System¶

# worker-controller/application/hosted_services/worker_reconciler.py

class WorkerReconciler(WatchTriggeredHostedService[CMLWorkerReadModel]):
    """Reconciles worker state; also runs discovery as an independent task."""

    async def _run_discovery_loop(self) -> None:
        """Discovery loop - scans AWS for EC2 instances matching AMI patterns."""
        while self._running:
            for region in self._discovery_regions:
                ami_ids = await ec2_client.get_ami_ids_by_name(ami_pattern)
                instances = await ec2_client.list_instances_by_ami(ami_ids)
                await self._api.bulk_import_workers(
                    discovered_instances=instances,
                    aws_region=region,
                    source="worker-controller-discovery",
                )
            await asyncio.sleep(self._discovery_interval)

# worker-controller/application/hosted_services/worker_reconciler.py

class WorkerReconciler(WatchTriggeredHostedService[CMLWorkerReadModel]):
    """Reconciles worker state with actual EC2/CML state."""

    async def reconcile(self, worker: CMLWorkerReadModel) -> ReconciliationResult:
        """Reconcile a single worker."""
        if worker.status == "RUNNING":
            return await self._handle_running(worker)
        # ... handle other states ...

    async def _handle_running(self, worker: CMLWorkerReadModel) -> ReconciliationResult:
        """Handle RUNNING worker - collect metrics and detect activity."""
        # 1. Collect EC2 status and CloudWatch metrics
        ec2_metrics = await self._ec2.get_instance_metrics(worker.ec2_instance_id)

        # 2. Collect CML system stats
        cml_metrics = await self._cml.get_system_stats(worker.ip_address)

        # 3. Report metrics to Control Plane API
        await self._api.report_worker_metrics(worker.id, {
            "collected_at": datetime.now(timezone.utc).isoformat(),
            "ec2": ec2_metrics,
            "cml": cml_metrics,
        })

        # 4. Detect activity and idle state
        await self._detect_activity(worker)

        return ReconciliationResult.success()

# control-plane-api/application/services/sse_event_relay.py

class SSEEventRelay:
    """Relay service for broadcasting events to SSE clients."""

    async def broadcast_event(self, event_type: str, data: dict) -> None:
        """Broadcast event to all matching clients."""
        # Publish to Redis if enabled (multi-instance support)
        if self._redis_client:
            await self._redis_client.publish(self.REDIS_CHANNEL, payload)
        else:
            await self._broadcast_local(event_message)

@dataclass
class DiscoverySettings:
    enabled: bool = True
    regions: list[str] = field(default_factory=lambda: ["us-east-1"])
    ami_name_pattern: str = "CML-*"
    scan_interval_seconds: int = 300

new_status = _map_ec2_state_to_cml_status(ec2_state)
# running -> RUNNING
# stopped -> STOPPED
# pending -> PENDING
# etc.

metrics = aws_client.get_instance_resources_utilization(
    aws_region=worker.state.aws_region,
    instance_id=worker.state.aws_instance_id,
    relative_start_time=FIVE_MIN_AGO
)

worker.update_telemetry(
    cpu_utilization=metrics.avg_cpu_utilization,
    memory_utilization=metrics.avg_memory_utilization,
    last_activity_at=datetime.now(timezone.utc)
)

metrics_data = {
    "worker_id": worker_id,
    "worker_name": worker.state.name,
    "status": new_status.value,
    "metrics": {
        "cpu_utilization": cpu_util,
        "memory_utilization": memory_util
    }
}

for observer in self._observers:
    observer(metrics_data)

# src/application/settings.py

class Settings(ApplicationSettings):
    # Worker Monitoring
    worker_monitoring_enabled: bool = True
    worker_metrics_poll_interval: int = 300  # 5 minutes
    worker_notification_webhooks: list[str] = []

    # Background Job Store
    background_job_store: dict[str, Any] = {
        "redis_host": "redis",
        "redis_port": 6379,
        "redis_db": 1
    }

# main.py

def configure_worker_monitoring(app: FastAPI) -> None:
    """Configure worker monitoring on startup"""

    # Get services from DI container
    worker_repository = app.state.services.get_required_service(CMLWorkerRepository)
    aws_client = app.state.services.get_required_service(AwsEc2Client)
    background_task_bus = app.state.services.get_required_service(BackgroundTasksBus)
    background_task_scheduler = app.state.services.get_required_service(BackgroundTaskScheduler)

    # Create notification handler
    notification_handler = WorkerNotificationHandler(
        cpu_threshold=90.0,
        memory_threshold=90.0
    )

    # Create monitoring scheduler
    scheduler = WorkerMonitoringScheduler(
        worker_repository=worker_repository,
        aws_client=aws_client,
        notification_handler=notification_handler,
        background_task_bus=background_task_bus,
        background_task_scheduler=background_task_scheduler,
        poll_interval=app_settings.worker_metrics_poll_interval
    )

    # Add lifecycle hooks
    @app.on_event("startup")
    async def start_monitoring():
        await scheduler.start_async()

    @app.on_event("shutdown")
    async def stop_monitoring():
        await scheduler.stop_async()

# Domain event triggers monitoring
@integration_event_handler(CMLWorkerCreatedDomainEvent)
async def on_worker_created(event: CMLWorkerCreatedDomainEvent):
    await monitoring_scheduler.start_monitoring_worker_async(event.aggregate_id)

# Job checks worker status
if worker.state.status == CMLWorkerStatus.TERMINATED:
    raise Exception("Worker terminated - stopping job")

# Scheduler removes job
await monitoring_scheduler.stop_monitoring_worker_async(worker_id)

# domain/entities/cml_worker.py

class CMLWorkerState:
    cpu_utilization: float | None
    memory_utilization: float | None
    active_labs_count: int
    last_activity_at: datetime | None

with tracer.start_as_current_span("collect_worker_metrics") as span:
    span.set_attribute("worker_id", worker_id)
    span.set_attribute("aws_instance_id", instance_id)
    span.set_attribute("cpu_utilization", cpu_util)

logger.info(f"📊 Collected metrics for worker {worker_id}: CPU={cpu}%, Memory={mem}%")
logger.warning(f"⚠️ HIGH CPU: Worker {worker_id} - {cpu}% (threshold: 90%)")
logger.error(f"❌ Failed to collect metrics for worker {worker_id}: {error}")

handler = WorkerNotificationHandler(
    cpu_threshold=90.0,    # Alert if CPU > 90%
    memory_threshold=90.0  # Alert if memory > 90%
)

def _handle_threshold_violation(
    self,
    worker_id: str,
    worker_name: str,
    metric_type: str,  # "cpu" or "memory"
    value: float,
    threshold: float
) -> None:
    logger.warning(
        f"🚨 Threshold Violation: {worker_name} - "
        f"{metric_type.upper()} at {value:.1f}% exceeds {threshold}%"
    )

    # TODO: Future integrations
    # - await self._send_webhook_notification(...)
    # - await self._trigger_pagerduty_alert(...)

worker = await worker_repository.get_by_id_async(worker_id)
if not worker:
    logger.warning(f"Worker {worker_id} not found - stopping job")
    raise Exception("Worker not found - terminating job")

try:
    status_checks = aws_client.get_instance_status_checks(...)
except Exception as e:
    logger.error(f"Failed to get EC2 status: {e}")
    # Job continues on next interval

def configure(self, service_provider=None, **kwargs):
    if service_provider:
        self.aws_ec2_client = service_provider.get_required_service(AwsEc2Client)
        self.worker_repository = service_provider.get_required_service(CMLWorkerRepository)

# Check EC2 API access
status = aws_client.get_instance_status_checks(...)

# Check CloudWatch metrics
metrics = aws_client.get_instance_resources_utilization(...)

# Increase poll interval to 10 minutes
worker_metrics_poll_interval: int = 600