Enh(aws): Incremental syncing to aws

Author: klieret

aws/docker_and_sync.sh

@@ -63,6 +63,8 @@ docker run hello-world
 # Pull images from ECR so we don't have to build them
 export AWS_DOCKER_REGISTRY="039984708918.dkr.ecr.us-east-1.amazonaws.com"
 aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin $AWS_DOCKER_REGISTRY
+export AWS_S3_BUCKET="codeclash"
+export AWS_S3_PREFIX="logs"
 
 # Create logs directory
 mkdir -p logs

codeclash/analysis/matrix.py

@@ -11,6 +11,7 @@
 from codeclash.constants import DIR_WORK
 from codeclash.games import get_game
 from codeclash.tournaments.utils.git_utils import filter_git_diff
+from codeclash.utils.atomic_write import atomic_write
 from codeclash.utils.log import add_file_handler, get_logger
 
 # todo: add visualization code
@@ -90,11 +91,7 @@ def matrices(self) -> dict:
     def _save(self):
         """Save metadata to matrix.json."""
         with self._save_lock:
-            # let's make this high stakes write atomic, because else if you run out of disk space,
-            # you'll lose all progress
-            tmp_file = self.output_file.with_suffix(".tmp")
-            tmp_file.write_text(json.dumps(self._metadata, indent=2))
-            tmp_file.rename(self.output_file)
+            atomic_write(self.output_file, json.dumps(self._metadata, indent=2))
 
     def _load_existing_progress(self):
         """Load existing progress from matrix.json if it exists."""

codeclash/tournaments/pvp.py

@@ -15,6 +15,8 @@
 from codeclash.games import get_game
 from codeclash.games.game import CodeGame
 from codeclash.tournaments.tournament import AbstractTournament
+from codeclash.utils.atomic_write import atomic_write
+from codeclash.utils.aws import is_running_in_aws_batch, s3_log_sync
 from codeclash.utils.environment import copy_to_container
 
 
@@ -127,7 +129,11 @@ def run_agent(self, agent: Player, round_num: int) -> None:
 
     def _save(self) -> None:
         self.local_output_dir.mkdir(parents=True, exist_ok=True)
-        (self.local_output_dir / "metadata.json").write_text(json.dumps(self.get_metadata(), indent=2))
+        metadata_file = self.local_output_dir / "metadata.json"
+        atomic_write(metadata_file, json.dumps(self.get_metadata(), indent=2))
+        self.logger.debug(f"Metadata saved to {metadata_file}")
+        if is_running_in_aws_batch():
+            s3_log_sync(self.local_output_dir, logger=self.logger)
 
     def _compress_round_logs(self) -> None:
         self.logger.info("Compressing round logs, this might take a while...")

codeclash/tournaments/single_player.py

@@ -15,6 +15,8 @@
 from codeclash.games.game import CodeGame
 from codeclash.tournaments.tournament import AbstractTournament
 from codeclash.tournaments.utils.git_utils import filter_git_diff
+from codeclash.utils.atomic_write import atomic_write
+from codeclash.utils.aws import is_running_in_aws_batch, s3_log_sync
 from codeclash.utils.environment import copy_to_container
 
 
@@ -131,7 +133,9 @@ def set_mirror_state_to_round(self, round_num: int):
 
     def _save(self) -> None:
         self.local_output_dir.mkdir(parents=True, exist_ok=True)
-        (self.local_output_dir / "metadata.json").write_text(json.dumps(self.get_metadata(), indent=2))
+        atomic_write(self.local_output_dir / "metadata.json", json.dumps(self.get_metadata(), indent=2))
+        if is_running_in_aws_batch():
+            s3_log_sync(self.local_output_dir, logger=self.logger)
 
     def end(self):
         """Clean up game resources."""

codeclash/utils/atomic_write.py

@@ -0,0 +1,8 @@
+from pathlib import Path
+
+
+def atomic_write(path: Path, text: str) -> None:
+    """Write text to file atomically to prevent corruption on interruption."""
+    tmp_file = path.with_suffix(".tmp")
+    tmp_file.write_text(text)
+    tmp_file.rename(path)

codeclash/utils/aws.py

@@ -1,6 +1,9 @@
 import os
 import subprocess
 from logging import Logger
+from pathlib import Path
+
+from codeclash.constants import DIR_LOGS
 
 
 def is_running_in_aws_batch() -> bool:
@@ -56,3 +59,45 @@ def pull_game_container_aws_ecr(*, game_name: str, image_name: str, logger: Logg
         raise RuntimeError(f"Failed to pull and tag Docker image: {result.stderr}")
 
     logger.info(f"✅ Pulled and tagged Docker image {image_name}")
+
+
+def s3_log_sync(local_output_dir: Path, *, logger: Logger) -> None:
+    """Sync local logs to S3 bucket.
+
+    Args:
+        local_output_dir: Local directory containing logs to sync
+        logger: Logger instance for debug output
+
+    Raises:
+        AssertionError: If required environment variables are not set
+        RuntimeError: If aws s3 sync fails
+    """
+    aws_s3_bucket = os.getenv("AWS_S3_BUCKET")
+    aws_s3_prefix = os.getenv("AWS_S3_PREFIX")
+
+    assert aws_s3_bucket is not None, "AWS_S3_BUCKET environment variable must be set"
+    assert aws_s3_prefix is not None, "AWS_S3_PREFIX environment variable must be set"
+
+    # Construct S3 path: s3://bucket/prefix/logs/relative_path
+    # where relative_path is local_output_dir relative to DIR_LOGS
+    try:
+        relative_path = local_output_dir.relative_to(DIR_LOGS)
+    except ValueError:
+        # If local_output_dir is not under DIR_LOGS, use the full path
+        relative_path = local_output_dir
+
+    s3_path = f"s3://{aws_s3_bucket}/{aws_s3_prefix}/logs/{relative_path}"
+
+    logger.debug(f"Syncing {local_output_dir} to {s3_path}")
+
+    result = subprocess.run(
+        ["aws", "s3", "sync", str(local_output_dir), s3_path, "--exclude", "*/rounds/*"],
+        capture_output=True,
+        text=True,
+    )
+
+    if result.returncode != 0:
+        logger.critical(f"❌ Failed to sync logs to S3: {result.stderr}\n{result.stdout}")
+        raise RuntimeError(f"Failed to sync logs to S3: {result.stderr}")
+
+    logger.info(f"✅ Successfully synced logs to {s3_path}")