From d79a8a52db1e0102aaccddaee6bdc5248bc229fb Mon Sep 17 00:00:00 2001 From: Obada Haddad Date: Fri, 3 Jul 2026 16:09:49 +0200 Subject: [PATCH 1/7] add option in compute worker env to not send logs to the instance, instead writing them in a local file --- compute_worker/compute_worker.py | 159 +++++++++++++++++++------------ 1 file changed, 100 insertions(+), 59 deletions(-) diff --git a/compute_worker/compute_worker.py b/compute_worker/compute_worker.py index 243dcd16d..d4268a0b0 100644 --- a/compute_worker/compute_worker.py +++ b/compute_worker/compute_worker.py @@ -113,6 +113,8 @@ def to_bool(val): WORKER_BUNDLE_URL_REWRITE = get("WORKER_BUNDLE_URL_REWRITE", "").strip() + SILENT_COMPUTE_WORKER = get("SILENT_COMPUTE_WORKER", "false").lower() + # ----------------------------------------------- # Program Kind @@ -331,9 +333,15 @@ def run_wrapper(run_args): except SubmissionException as e: msg = str(e).strip() if msg: - msg = f"Submission failed: {msg}. See logs for more details." + if Settings.SILENT_COMPUTE_WORKER == "true": + msg = f"Submission failed: {msg}. Contact the Organizer for more details." + else: + msg = f"Submission failed: {msg}. See logs for more details." else: - msg = "Submission failed. See logs for more details." + if Settings.SILENT_COMPUTE_WORKER == "true": + msg = "Submission failed. Contact the Organizer for more details." + else: + msg = "Submission failed. See logs for more details." run._update_status(SubmissionStatus.FAILED, extra_information=msg) raise except Exception as e: @@ -546,18 +554,36 @@ async def watch_detailed_results(self): def push_logs(self): """Upload any collected logs, even in case of crash. """ - try: - for kind, logs in (self.logs or {}).items(): - for stream_key in ("stdout", "stderr"): - entry = logs.get(stream_key) if isinstance(logs, dict) else None - if not entry: - continue - location = entry.get("location") - data = entry.get("data") or b"" - if location: - self._put_file(location, raw_data=data) - except Exception as e: - logger.exception(f"Failed best-effort log upload: {e}") + if Settings.SILENT_COMPUTE_WORKER == "false": + try: + for kind, logs in (self.logs or {}).items(): + for stream_key in ("stdout", "stderr"): + entry = logs.get(stream_key) if isinstance(logs, dict) else None + if not entry: + continue + location = entry.get("location") + data = entry.get("data") or b"" + if location: + self._put_file(location, raw_data=data) + except Exception as e: + logger.exception(f"Failed best-effort log upload: {e}") + + else: + try: + logs_path = os.path.join(self.root_dir, "logs") + with open(logs_path, "w") as f: + for kind, logs in (self.logs or {}).items(): + for stream_key in ("stdout", "stderr"): + entry = logs.get(stream_key) if isinstance(logs, dict) else None + if not entry: + continue + location = entry.get("location") + data = entry.get("data") or b"" + if location: + f.write(str(data)) + except Exception as e: + logger.exception(f"Failed best-effort log file creation: {e}") + def get_detailed_results_file_path(self): default_detailed_results_path = os.path.join( @@ -671,9 +697,14 @@ def _get_container_image(self, image_name): self._update_submission(docker_pull_fail_data) # Send error through web socket to the frontend asyncio.run(self._send_data_through_socket(str(pull_error))) - raise DockerImagePullException( - f"Pull for {image_name} failed! Check the logs for more information" - ) + if Settings.SILENT_COMPUTE_WORKER == "true": + raise DockerImagePullException( + f"Pull for {image_name} failed! Contact the Organizer for more details." + ) + else: + raise DockerImagePullException( + f"Pull for {image_name} failed! Check the logs for more information" + ) else: logger.warning("Failed. Retrying in 5 seconds...") time.sleep(5) # Wait 5 seconds before retrying @@ -886,21 +917,24 @@ async def _run_container_engine_cmd(self, container, kind): # Create a websocket to send the logs in real time to the codabench instance # We need to set a timeout for the websocket connection otherwise the program will get stuck if he websocket does not connect. websocket = None - try: - websocket_url = f"{self.websocket_url}?kind={kind}" - logger.debug(f"Connecting to {websocket_url} for container {str(container.get('Id'))}") - websocket = await asyncio.wait_for( - websockets.connect(websocket_url), timeout=10.0 - ) - logger.debug(f"connected to {websocket_url} for container {str(container.get('Id'))}") - except Exception as e: - logger.error( - f"There was an error trying to connect to the websocket on the codabench instance: {e}" - ) + # Do not create a websocket if the real time logs are not wanted (Silent Compute Worker) + if Settings.SILENT_COMPUTE_WORKER == "false": + try: + websocket_url = f"{self.websocket_url}?kind={kind}" + logger.debug(f"Connecting to {websocket_url} for container {str(container.get('Id'))}") + websocket = await asyncio.wait_for( + websockets.connect(websocket_url), timeout=10.0 + ) + logger.debug(f"connected to {websocket_url} for container {str(container.get('Id'))}") - if Settings.LOG_LEVEL == Settings.LOG_LEVEL_DEBUG: - logger.exception(e) + except Exception as e: + logger.error( + f"There was an error trying to connect to the websocket on the codabench instance: {e}" + ) + + if Settings.LOG_LEVEL == Settings.LOG_LEVEL_DEBUG: + logger.exception(e) start = time.time() @@ -916,6 +950,7 @@ async def _run_container_engine_cmd(self, container, kind): ) # If we enter the for loop after the container exited, the program will get stuck + # Do not send the real time logs if they are not wanted (Silent Compute Worker) if client.inspect_container(container)["State"]["Status"].lower() == "running": logger.debug( "Show the logs and stream them to codabench " + container.get("Id") @@ -925,25 +960,27 @@ async def _run_container_engine_cmd(self, container, kind): if log[0] is not None: stdout_chunks.append(log[0]) logger.info(log[0].decode()) - try: - if websocket is not None: - await websocket.send( - json.dumps({"kind": kind, "message": log[0].decode()}) - ) - except Exception as e: - logger.error(e) + if Settings.SILENT_COMPUTE_WORKER == "false": + try: + if websocket is not None: + await websocket.send( + json.dumps({"kind": kind, "message": log[0].decode()}) + ) + except Exception as e: + logger.error(e) # Errors elif log[1] is not None: stderr_chunks.append(log[1]) logger.error(log[1].decode()) - try: - if websocket is not None: - await websocket.send( - json.dumps({"kind": kind, "message": log[1].decode()}) - ) - except Exception as e: - logger.error(e) + if Settings.SILENT_COMPUTE_WORKER == "false": + try: + if websocket is not None: + await websocket.send( + json.dumps({"kind": kind, "message": log[1].decode()}) + ) + except Exception as e: + logger.error(e) except (docker.errors.NotFound, docker.errors.APIError) as e: logger.error(e) @@ -959,15 +996,17 @@ async def _run_container_engine_cmd(self, container, kind): # Gets the logs of the container, sperating stdout and stderr (first and second position) thanks for demux=True return_Code = client.wait(container) logs_Unified = (b"".join(stdout_chunks), b"".join(stderr_chunks)) - logger.debug( - f"WORKER_MARKER: Disconnecting from {websocket_url}, program counter = {self.completed_program_counter}" - ) - if websocket is not None: - try: - await websocket.close() - await websocket.wait_closed() - except Exception as e: - logger.error(e) + + if Settings.SILENT_COMPUTE_WORKER == "false": + logger.debug( + f"WORKER_MARKER: Disconnecting from {websocket_url}, program counter = {self.completed_program_counter}" + ) + if websocket is not None: + try: + await websocket.close() + await websocket.wait_closed() + except Exception as e: + logger.error(e) client.remove_container(container, v=True, force=True) logger.debug(f"Container {container.get('Id')} exited with status code : {str(return_Code['StatusCode'])}") @@ -1414,12 +1453,14 @@ def start(self): self.ingestion_program_exit_code = return_code self.ingestion_program_elapsed_time = elapsed_time logger.info(f"[exited with {logs['returncode']}]") - for key, value in logs.items(): - if key not in ["stdout", "stderr"]: - continue - if value["data"]: - logger.info(f"[{key}]\n{value['data']}") - self._put_file(value["location"], raw_data=value["data"]) + if Settings.SILENT_COMPUTE_WORKER == "false": + for key, value in logs.items(): + if key not in ["stdout", "stderr"]: + continue + if value["data"]: + logger.info(f"[{key}]\n{value['data']}") + self._put_file(value["location"], raw_data=value["data"]) + # set logs of this kind to None, since we handled them already logger.info("Program finished") From ca1f03c61e6dc308e8db41ffee003959b408723a Mon Sep 17 00:00:00 2001 From: Obada Haddad Date: Fri, 3 Jul 2026 16:18:16 +0200 Subject: [PATCH 2/7] rename the No Cleanup env variable, add documentation --- compute_worker/compute_worker.py | 6 +++--- docker-compose.yml | 2 +- .../Compute-Worker-Management---Setup.md | 6 ++++++ 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/compute_worker/compute_worker.py b/compute_worker/compute_worker.py index d4268a0b0..baafca700 100644 --- a/compute_worker/compute_worker.py +++ b/compute_worker/compute_worker.py @@ -109,7 +109,7 @@ def to_bool(val): COMPETITION_CONTAINER_HTTP_PROXY = get("COMPETITION_CONTAINER_HTTP_PROXY", "") COMPETITION_CONTAINER_HTTPS_PROXY = get("COMPETITION_CONTAINER_HTTPS_PROXY", "") - CODALAB_IGNORE_CLEANUP_STEP = to_bool(get("CODALAB_IGNORE_CLEANUP_STEP")) + COMPUTE_WORKER_NO_CLEANUP = to_bool(get("COMPUTE_WORKER_NO_CLEANUP")) WORKER_BUNDLE_URL_REWRITE = get("WORKER_BUNDLE_URL_REWRITE", "").strip() @@ -1563,9 +1563,9 @@ def push_output(self): self._put_dir(self.scoring_result, self.output_dir) def clean_up(self): - if Settings.CODALAB_IGNORE_CLEANUP_STEP: + if Settings.COMPUTE_WORKER_NO_CLEANUP: logger.warning( - f"CODALAB_IGNORE_CLEANUP_STEP mode enabled, ignoring clean up of: {self.root_dir}" + f"COMPUTE_WORKER_NO_CLEANUP mode enabled, ignoring clean up of: {self.root_dir}" ) return diff --git a/docker-compose.yml b/docker-compose.yml index 5b66662c7..d4458f70d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -244,7 +244,7 @@ services: environment: - BROKER_URL=pyamqp://${RABBITMQ_DEFAULT_USER}:${RABBITMQ_DEFAULT_PASS}@${RABBITMQ_HOST}:${RABBITMQ_PORT}// # Make the worker leave behind the submission so we can examine it - - CODALAB_IGNORE_CLEANUP_STEP=1 + - COMPUTE_WORKER_NO_CLEANUP="true" tty: true logging: options: diff --git a/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md b/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md index 5735af914..2e761b613 100644 --- a/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md +++ b/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md @@ -59,6 +59,12 @@ HOST_DIRECTORY=/codabench CONTAINER_ENGINE_EXECUTABLE=docker #USE_GPU=True #GPU_DEVICE=nvidia.com/gpu=all +# This option removes the ability of the compute worker to send logs to +# codabench, instead writing them locally on disk. Combine with +# COMPUTE_WORKER_NO_CLEANUP=true to stop the worker's cleanup to keep +# all the logs locally only +#SILENT_COMPUTE_WORKER=False +#COMPUTE_WORKER_NO_CLEANUP=false ####################################################################### # Network # From b828447519fb5f0968b4128834d3e4bdb39338de Mon Sep 17 00:00:00 2001 From: Obada Haddad Date: Mon, 6 Jul 2026 15:53:50 +0200 Subject: [PATCH 3/7] use real boolean values --- compute_worker/compute_worker.py | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/compute_worker/compute_worker.py b/compute_worker/compute_worker.py index baafca700..a8284ce94 100644 --- a/compute_worker/compute_worker.py +++ b/compute_worker/compute_worker.py @@ -109,11 +109,11 @@ def to_bool(val): COMPETITION_CONTAINER_HTTP_PROXY = get("COMPETITION_CONTAINER_HTTP_PROXY", "") COMPETITION_CONTAINER_HTTPS_PROXY = get("COMPETITION_CONTAINER_HTTPS_PROXY", "") - COMPUTE_WORKER_NO_CLEANUP = to_bool(get("COMPUTE_WORKER_NO_CLEANUP")) + COMPUTE_WORKER_NO_CLEANUP = to_bool(get("COMPUTE_WORKER_NO_CLEANUP", "False")) WORKER_BUNDLE_URL_REWRITE = get("WORKER_BUNDLE_URL_REWRITE", "").strip() - SILENT_COMPUTE_WORKER = get("SILENT_COMPUTE_WORKER", "false").lower() + SILENT_COMPUTE_WORKER = to_bool(get("SILENT_COMPUTE_WORKER", "False")) # ----------------------------------------------- @@ -333,12 +333,12 @@ def run_wrapper(run_args): except SubmissionException as e: msg = str(e).strip() if msg: - if Settings.SILENT_COMPUTE_WORKER == "true": + if Settings.SILENT_COMPUTE_WORKER: msg = f"Submission failed: {msg}. Contact the Organizer for more details." else: msg = f"Submission failed: {msg}. See logs for more details." else: - if Settings.SILENT_COMPUTE_WORKER == "true": + if Settings.SILENT_COMPUTE_WORKER: msg = "Submission failed. Contact the Organizer for more details." else: msg = "Submission failed. See logs for more details." @@ -554,7 +554,7 @@ async def watch_detailed_results(self): def push_logs(self): """Upload any collected logs, even in case of crash. """ - if Settings.SILENT_COMPUTE_WORKER == "false": + if Settings.SILENT_COMPUTE_WORKER == False: try: for kind, logs in (self.logs or {}).items(): for stream_key in ("stdout", "stderr"): @@ -697,7 +697,7 @@ def _get_container_image(self, image_name): self._update_submission(docker_pull_fail_data) # Send error through web socket to the frontend asyncio.run(self._send_data_through_socket(str(pull_error))) - if Settings.SILENT_COMPUTE_WORKER == "true": + if Settings.SILENT_COMPUTE_WORKER: raise DockerImagePullException( f"Pull for {image_name} failed! Contact the Organizer for more details." ) @@ -919,7 +919,7 @@ async def _run_container_engine_cmd(self, container, kind): websocket = None # Do not create a websocket if the real time logs are not wanted (Silent Compute Worker) - if Settings.SILENT_COMPUTE_WORKER == "false": + if Settings.SILENT_COMPUTE_WORKER == False: try: websocket_url = f"{self.websocket_url}?kind={kind}" logger.debug(f"Connecting to {websocket_url} for container {str(container.get('Id'))}") @@ -960,7 +960,7 @@ async def _run_container_engine_cmd(self, container, kind): if log[0] is not None: stdout_chunks.append(log[0]) logger.info(log[0].decode()) - if Settings.SILENT_COMPUTE_WORKER == "false": + if Settings.SILENT_COMPUTE_WORKER == False: try: if websocket is not None: await websocket.send( @@ -973,7 +973,7 @@ async def _run_container_engine_cmd(self, container, kind): elif log[1] is not None: stderr_chunks.append(log[1]) logger.error(log[1].decode()) - if Settings.SILENT_COMPUTE_WORKER == "false": + if Settings.SILENT_COMPUTE_WORKER == False: try: if websocket is not None: await websocket.send( @@ -997,7 +997,7 @@ async def _run_container_engine_cmd(self, container, kind): return_Code = client.wait(container) logs_Unified = (b"".join(stdout_chunks), b"".join(stderr_chunks)) - if Settings.SILENT_COMPUTE_WORKER == "false": + if Settings.SILENT_COMPUTE_WORKER == False: logger.debug( f"WORKER_MARKER: Disconnecting from {websocket_url}, program counter = {self.completed_program_counter}" ) @@ -1356,6 +1356,7 @@ def start(self): self._run_program_directory(kind=ProgramKind.INGESTION_PROGRAM, program_dir=ingestion_program_dir), ]) + logger.info(tasks) gathered_tasks = asyncio.gather(*tasks, return_exceptions=True) task_results = [] # will store results/exceptions from gather @@ -1453,7 +1454,7 @@ def start(self): self.ingestion_program_exit_code = return_code self.ingestion_program_elapsed_time = elapsed_time logger.info(f"[exited with {logs['returncode']}]") - if Settings.SILENT_COMPUTE_WORKER == "false": + if Settings.SILENT_COMPUTE_WORKER == False: for key, value in logs.items(): if key not in ["stdout", "stderr"]: continue From 8adf8c0fa008953f797543316a1a917055764802 Mon Sep 17 00:00:00 2001 From: Obada Haddad Date: Mon, 6 Jul 2026 16:27:44 +0200 Subject: [PATCH 4/7] docs: update value to use Boolean False --- .../Running_a_benchmark/Compute-Worker-Management---Setup.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md b/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md index 2e761b613..57781f5ca 100644 --- a/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md +++ b/documentation/docs/Organizers/Running_a_benchmark/Compute-Worker-Management---Setup.md @@ -64,7 +64,7 @@ CONTAINER_ENGINE_EXECUTABLE=docker # COMPUTE_WORKER_NO_CLEANUP=true to stop the worker's cleanup to keep # all the logs locally only #SILENT_COMPUTE_WORKER=False -#COMPUTE_WORKER_NO_CLEANUP=false +#COMPUTE_WORKER_NO_CLEANUP=False ####################################################################### # Network # From 9fed924761eb1611a08854b501bf33a00402b79d Mon Sep 17 00:00:00 2001 From: Obada Haddad Date: Mon, 6 Jul 2026 16:50:20 +0200 Subject: [PATCH 5/7] update logs_loguru to inclue new tasks variable names to color them --- src/settings/logs_loguru.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/settings/logs_loguru.py b/src/settings/logs_loguru.py index 28b2cf075..137ff83c9 100644 --- a/src/settings/logs_loguru.py +++ b/src/settings/logs_loguru.py @@ -130,7 +130,17 @@ def colorize_run_args(json_str): json_str, ) json_str = re.sub( - r'("ingestion_program": ")(.*?)(",)', + r'("ingestion_program_data": ")(.*?)(",)', + rf"\1{yellow}\2{reset}\3{lineskip}", + json_str, + ) + json_str = re.sub( + r'("submission_data": ")(.*?)(",)', + rf"\1{yellow}\2{reset}\3{lineskip}", + json_str, + ) + json_str = re.sub( + r'("scoring_program_data": ")(.*?)(",)', rf"\1{yellow}\2{reset}\3{lineskip}", json_str, ) From 2f29f1e81dc2ad35749f3967b1c9c186935cd1ed Mon Sep 17 00:00:00 2001 From: Obada Haddad Date: Mon, 6 Jul 2026 16:51:04 +0200 Subject: [PATCH 6/7] change variable name to use boolean --- docker-compose.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker-compose.yml b/docker-compose.yml index d4458f70d..0bd754e22 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -244,7 +244,7 @@ services: environment: - BROKER_URL=pyamqp://${RABBITMQ_DEFAULT_USER}:${RABBITMQ_DEFAULT_PASS}@${RABBITMQ_HOST}:${RABBITMQ_PORT}// # Make the worker leave behind the submission so we can examine it - - COMPUTE_WORKER_NO_CLEANUP="true" + - COMPUTE_WORKER_NO_CLEANUP=True tty: true logging: options: From 5bb39b21f46649116305baf54ea22f7bebd84b41 Mon Sep 17 00:00:00 2001 From: Obada Haddad Date: Tue, 7 Jul 2026 14:44:15 +0200 Subject: [PATCH 7/7] fix some syntax --- compute_worker/compute_worker.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/compute_worker/compute_worker.py b/compute_worker/compute_worker.py index a8284ce94..c08c4139c 100644 --- a/compute_worker/compute_worker.py +++ b/compute_worker/compute_worker.py @@ -334,12 +334,12 @@ def run_wrapper(run_args): msg = str(e).strip() if msg: if Settings.SILENT_COMPUTE_WORKER: - msg = f"Submission failed: {msg}. Contact the Organizer for more details." + msg = f"Submission failed: {msg}. Contact the Organizer(s) for more details." else: msg = f"Submission failed: {msg}. See logs for more details." else: if Settings.SILENT_COMPUTE_WORKER: - msg = "Submission failed. Contact the Organizer for more details." + msg = "Submission failed. Contact the Organizer(s) for more details." else: msg = "Submission failed. See logs for more details." run._update_status(SubmissionStatus.FAILED, extra_information=msg) @@ -699,7 +699,7 @@ def _get_container_image(self, image_name): asyncio.run(self._send_data_through_socket(str(pull_error))) if Settings.SILENT_COMPUTE_WORKER: raise DockerImagePullException( - f"Pull for {image_name} failed! Contact the Organizer for more details." + f"Pull for {image_name} failed! Contact the Organizer(s) for more details." ) else: raise DockerImagePullException( @@ -843,7 +843,8 @@ def _create_container( "SYS_CHROOT", ] - # Configure whether or not we use the GPU. Also setting auto_remove to False because + # Configure whether or not we use the GPU. Also setting auto_remove to False because removing too fast + # can bug out the worker (can't get the logs fast enough) if Settings.CONTAINER_ENGINE_EXECUTABLE == Settings.DOCKER: security_options = ["no-new-privileges"] else: