Monitor AWS Batch jobs
| 20 | |
| 21 | |
| 22 | class AWSBatchMonitor: |
| 23 | """Monitor AWS Batch jobs""" |
| 24 | |
| 25 | def __init__( |
| 26 | self, |
| 27 | *, |
| 28 | job_queue: str = "codeclash-queue", |
| 29 | region: str = "us-east-1", |
| 30 | logs_base_dir: Path | None = None, |
| 31 | ): |
| 32 | if boto3 is None: |
| 33 | msg = "boto3 is not installed. Install it with: pip install codeclash[aws]" |
| 34 | raise ImportError(msg) |
| 35 | self.batch_client = boto3.client("batch", region_name=region) |
| 36 | self.job_queue = job_queue |
| 37 | self.region = region |
| 38 | self.logs_base_dir = logs_base_dir or Path("logs") |
| 39 | self._job_id_to_folder: dict[str, str] | None = None |
| 40 | self._job_id_to_round_info: dict[str, tuple[int, int] | None] | None = None |
| 41 | self._job_id_to_aws_command: dict[str, str | None] | None = None |
| 42 | |
| 43 | def list_jobs(self, *, limit: int | None = None, hours_back: int = 24) -> list[dict[str, Any]]: |
| 44 | """List all jobs from AWS Batch |
| 45 | |
| 46 | Args: |
| 47 | limit: Maximum number of jobs to return |
| 48 | hours_back: Number of hours to look back (default 24) |
| 49 | """ |
| 50 | all_jobs = [] |
| 51 | statuses = ["SUBMITTED", "PENDING", "RUNNABLE", "STARTING", "RUNNING", "SUCCEEDED", "FAILED"] |
| 52 | cutoff_timestamp = (datetime.now().timestamp() - (hours_back * 3600)) * 1000 |
| 53 | |
| 54 | for job_status in statuses: |
| 55 | try: |
| 56 | paginator = self.batch_client.get_paginator("list_jobs") |
| 57 | page_iterator = paginator.paginate(jobQueue=self.job_queue, jobStatus=job_status) |
| 58 | |
| 59 | for page in page_iterator: |
| 60 | jobs_in_page = page.get("jobSummaryList", []) |
| 61 | all_jobs.extend(jobs_in_page) |
| 62 | |
| 63 | except Exception as e: |
| 64 | logger.warning(f"Failed to list jobs with status {job_status}: {e}", exc_info=True) |
| 65 | |
| 66 | # Filter by time range |
| 67 | all_jobs = [job for job in all_jobs if job.get("createdAt", 0) >= cutoff_timestamp] |
| 68 | |
| 69 | all_jobs.sort(key=lambda x: x.get("createdAt", 0), reverse=True) |
| 70 | |
| 71 | if limit: |
| 72 | all_jobs = all_jobs[:limit] |
| 73 | |
| 74 | return all_jobs |
| 75 | |
| 76 | def format_job_for_display(self, job: dict[str, Any]) -> dict[str, Any]: |
| 77 | """Format job data for display in the UI""" |
| 78 | job_id = job.get("jobId", "") |
| 79 | job_name = job.get("jobName", "") |
no outgoing calls
no test coverage detected