diff --git a/Makefile b/Makefile index c29963fb..ea53010b 100644 --- a/Makefile +++ b/Makefile @@ -124,7 +124,7 @@ orchestration: nightly-worker-job nightly-manager-job compress_buckets metadata: tag-questions validate-questions -resolve: resolve-forecasts +resolve: resolve-forecasts push-resolution-sets leaderboards: leaderboard-tournament leaderboard-baseline leaderboard-preliminary @@ -222,6 +222,9 @@ validate-questions: resolve-forecasts: $(MAKE) -C src/orchestration/func_resolve || echo "* $@" >> $(MAKE_FAILURE_LOG) +push-resolution-sets: + $(MAKE) -C src/orchestration/func_push_resolution_sets || echo "* $@" >> $(MAKE_FAILURE_LOG) + naive-and-dummy-forecasters: $(MAKE) -C src/base_eval/naive_and_dummy_forecasters || echo "* $@" >> $(MAKE_FAILURE_LOG) diff --git a/src/helpers/git.py b/src/helpers/git.py index d6d8b6f3..9944bad6 100644 --- a/src/helpers/git.py +++ b/src/helpers/git.py @@ -57,7 +57,7 @@ def clone_and_push_files( files: Dict[str, str], commit_message: str, mirrors: Optional[List[str]] = None, -) -> None: +) -> bool: """Clone a Git repository, add/update files, commit, and push to origin and optional mirrors. Args: @@ -68,7 +68,9 @@ def clone_and_push_files( If None, attempts to load from secrets. Returns: - None. Exits with status 1 if an error is encountered while pushing. + bool: True if a new commit was created, False if HEAD was unchanged. + Origin and mirrors are pushed either way. Exits with status 1 if an + error is encountered while pushing. """ if not mirrors: mirrors = keys.get_secret_that_may_not_exist("HUGGING_FACE_REPO_URL") @@ -84,19 +86,26 @@ def clone_and_push_files( shutil.copy(source, full_destination_path, follow_symlinks=False) repo.index.add([destination]) + # Skip empty commits, but always push so a lagging mirror catches up. + has_changes = bool(repo.index.diff(repo.head.commit)) + if not has_changes: + logger.info(f"No new commit for {repo_url}; syncing remotes to HEAD.") + error_encountered = False author = Actor("ForecastBench bot", constants.BENCHMARK_EMAIL) committer = Actor("ForecastBench bot", constants.BENCHMARK_EMAIL) ssh_env = {"GIT_SSH_COMMAND": f"ssh -i {tmp_key_file_path} -o StrictHostKeyChecking=no"} try: - repo.index.commit(commit_message, author=author, committer=committer) + if has_changes: + repo.index.commit(commit_message, author=author, committer=committer) origin = repo.remote(name="origin") - origin.push(env=ssh_env) + # A rejected push does not raise in GitPython; surface it explicitly. + origin.push(env=ssh_env).raise_if_error() for index, mirror_url in enumerate(mirrors): mirror = repo.create_remote(f"mirror_{index}", url=mirror_url) - mirror.push(env=ssh_env) + mirror.push(env=ssh_env).raise_if_error() repo.delete_remote(mirror.name) - logger.info(f"Pushed to {mirror_url} (mirror) with commit message: {commit_message}") + logger.info(f"Pushed to {mirror_url} (mirror)") except Exception as e: error_encountered = True message = e.message if hasattr(e, "message") else str(e) @@ -108,14 +117,16 @@ def clone_and_push_files( if error_encountered: sys.exit(1) - logger.info(f"Pushed to {repo_url} with commit message: {commit_message}") + if has_changes: + logger.info(f"Pushed to {repo_url} with commit message: {commit_message}") + return has_changes def clone_commit_and_push( files: Dict[str, str], commit_message: str, ) -> None: - """Upload files files to Cloud Storage and push updates to Git. + """Push files to the git dataset repository. Args: files (Dict[str, str]): Mapping of local file paths to their git location. diff --git a/src/nightly_update_workflow/manager/main.py b/src/nightly_update_workflow/manager/main.py index c6835a6f..dc608f94 100644 --- a/src/nightly_update_workflow/manager/main.py +++ b/src/nightly_update_workflow/manager/main.py @@ -181,7 +181,7 @@ def main(): dict_to_use=dict_to_use_naive_and_dummy_forecasters, task_count=1 ) - # Block on resolve forecasts before launching leaderboards + # Block on resolve forecasts before pushing resolution sets to git cloud_run.block_and_check_job_result( operation=operation_resolve_forecasts, name=dict_to_use_resolve_forecasts, @@ -189,6 +189,19 @@ def main(): timeout=timeout_resolve_forecasts, ) + # Push all resolution sets in a single dedicated job now that the parallel resolve tasks have + # finished, so only one process pushes to the dataset repo (avoids a race condition). + dict_to_use_push_resolution_sets = "push_resolution_sets" + operation_push_resolution_sets = call_worker( + dict_to_use=dict_to_use_push_resolution_sets, + task_count=1, + ) + cloud_run.block_and_check_job_result( + operation=operation_push_resolution_sets, + name=dict_to_use_push_resolution_sets, + exit_on_error=False, + ) + # Launch leaderboard jobs in parallel dict_to_use_leaderboards = "leaderboards" timeout_leaderboards = cloud_run.timeout_1h * 4 diff --git a/src/nightly_update_workflow/worker/main.py b/src/nightly_update_workflow/worker/main.py index a0746b4f..71a111e1 100644 --- a/src/nightly_update_workflow/worker/main.py +++ b/src/nightly_update_workflow/worker/main.py @@ -17,6 +17,15 @@ ("func-resolve-forecasts", True, cloud_run.timeout_1h * 3, 50), ] ] + + +push_resolution_sets = [ + [ + ("func-push-resolution-sets", True, cloud_run.timeout_1h, 1), + ] +] + + leaderboards = [ [ ("func-leaderboard-tournament", True, cloud_run.timeout_1h * 4, 1), @@ -139,7 +148,8 @@ def main(): Env variables: CLOUD_RUN_TASK_INDEX: automatically set by Cloud Run Jobs - DICT_TO_USE: one of `fetch_and_update`, `metadata`, `resolve_forecasts`, `leaderboards`. + DICT_TO_USE: one of `fetch_and_update`, `metadata`, `resolve_forecasts`, + `push_resolution_sets`, `leaderboards`. """ dict_mapping = { "fetch_and_update": get_fetch_and_update(), @@ -147,6 +157,7 @@ def main(): "create_question_set": get_create_question_set(), "publish_question_set_make_llm_baseline": get_publish_question_set_make_llm_baseline(), "resolve_forecasts": resolve_forecasts, + "push_resolution_sets": push_resolution_sets, "leaderboards": leaderboards, "naive_and_dummy_forecasters": get_naive_and_dummy_forecasters(), "website": website, diff --git a/src/orchestration/_io.py b/src/orchestration/_io.py index c7d6cc8a..de3519ff 100644 --- a/src/orchestration/_io.py +++ b/src/orchestration/_io.py @@ -233,10 +233,11 @@ def upload_hash_mapping(raw_json: str, source_name: str) -> None: def upload_resolution_set(df: pd.DataFrame, forecast_due_date: str, question_set_filename: str): - """Upload resolution set to GCS and push to git.""" - from helpers import git # noqa: E402 - from helpers import keys # noqa: E402 + """Upload resolution set to GCS. + Only uploads to the bucket; resolution sets are pushed to git later by + `push_all_resolution_sets`. + """ basename = f"{forecast_due_date}_resolution_set.json" local_filename = f"/tmp/{basename}" df = df[["id", "source", "direction", "resolution_date", "resolved_to", "resolved"]] @@ -259,14 +260,52 @@ def upload_resolution_set(df: pd.DataFrame, forecast_due_date: str, question_set ) logger.info(f"Uploaded Resolution File {local_filename} to {upload_folder}.") + +def push_all_resolution_sets() -> None: + """Push every resolution set in the bucket to the git dataset repo in a single commit. + + Run as one job rather than having each parallel resolve task push its own set, which + avoids the race condition of concurrent pushes to the repository. + """ + from helpers import git, keys # noqa: E402 + + if env.RUNNING_LOCALLY: + logger.info("Running locally; not pushing resolution sets to git.") + return + + folder = "datasets/resolution_sets" + blob_names = gcp.storage.list_with_prefix( + bucket_name=env.PUBLIC_RELEASE_BUCKET, + prefix=f"{folder}/", + ) + + files = {} + for blob_name in blob_names: + basename = os.path.basename(blob_name) + if not basename.endswith("_resolution_set.json"): + continue + local_filename = f"/tmp/{basename}" + gcp.storage.download( + bucket_name=env.PUBLIC_RELEASE_BUCKET, + filename=blob_name, + local_filename=local_filename, + ) + files[local_filename] = f"{folder}/{basename}" + + if not files: + logger.warning(f"No resolution sets found under {folder}; nothing to push.") + return + mirrors = keys.get_secret_that_may_not_exist("HUGGING_FACE_REPO_URL") mirrors = [mirrors] if mirrors else [] - git.clone_and_push_files( + committed = git.clone_and_push_files( repo_url=keys.API_GITHUB_DATASET_REPO_URL, - files={local_filename: f"{upload_folder}/{basename}"}, - commit_message=f"resolution set: automatic update for {question_set_filename}.", + files=files, + commit_message="resolution sets: automatic update.", mirrors=mirrors, ) + if committed: + logger.info(f"Pushed {len(files)} resolution sets to git in a single commit.") def upload_processed_forecast_file(data: dict, forecast_due_date: str, filename: str): diff --git a/src/orchestration/func_push_resolution_sets/Makefile b/src/orchestration/func_push_resolution_sets/Makefile new file mode 100644 index 00000000..75786745 --- /dev/null +++ b/src/orchestration/func_push_resolution_sets/Makefile @@ -0,0 +1,37 @@ +all : + $(MAKE) clean + $(MAKE) deploy + +.PHONY : all clean deploy + +UPLOAD_DIR = upload +ROOT_DIR ?= $(abspath ../../..)/ +include $(ROOT_DIR)orchestration_upload.mk + +.gcloudignore: + cp -r $(ROOT_DIR)src/helpers/.gcloudignore . + +Dockerfile: $(ROOT_DIR)src/helpers/Dockerfile.template + sed \ + -e 's/REGION/$(CLOUD_DEPLOY_REGION)/g' \ + -e 's/STACK/google-22-full/g' \ + -e 's/PYTHON_VERSION/python312/g' \ + $< > Dockerfile + +deploy : main.py .gcloudignore requirements.txt Dockerfile + $(stage-orchestration-upload) + gcloud run jobs deploy \ + func-push-resolution-sets \ + --project $(CLOUD_PROJECT) \ + --region $(CLOUD_DEPLOY_REGION) \ + --tasks 1 \ + --parallelism 1 \ + --task-timeout 1h \ + --memory 2Gi \ + --max-retries 0 \ + --service-account $(QUESTION_BANK_BUCKET_SERVICE_ACCOUNT) \ + --set-env-vars $(DEFAULT_CLOUD_FUNCTION_ENV_VARS) \ + --source $(UPLOAD_DIR) + +clean : + rm -rf $(UPLOAD_DIR) .gcloudignore Dockerfile diff --git a/src/orchestration/func_push_resolution_sets/main.py b/src/orchestration/func_push_resolution_sets/main.py new file mode 100644 index 00000000..be370306 --- /dev/null +++ b/src/orchestration/func_push_resolution_sets/main.py @@ -0,0 +1,23 @@ +"""Cloud Run job: push all resolution sets to the git dataset repo in a single commit. + +See `orchestration._io.push_all_resolution_sets`. +""" + +import logging +from typing import Any + +from helpers import decorator +from orchestration import _io + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +@decorator.log_runtime +def driver(_: Any) -> None: + """Push all resolution sets to git in a single commit.""" + _io.push_all_resolution_sets() + + +if __name__ == "__main__": + driver(None) diff --git a/src/orchestration/func_push_resolution_sets/requirements.txt b/src/orchestration/func_push_resolution_sets/requirements.txt new file mode 100644 index 00000000..23e06c69 --- /dev/null +++ b/src/orchestration/func_push_resolution_sets/requirements.txt @@ -0,0 +1,10 @@ +google-cloud-storage +google-cloud-secret-manager +pandas>=2.2.2,<3.0 +GitPython +termcolor +slack_sdk +pandera +pytz +python-dateutil +backoff