Rebase #1154

.circleci/config.yml

@@ -3,7 +3,7 @@ version: 2
 jobs:
   test:
     machine:
-      image: ubuntu-2004:2022.07.1
+      image: ubuntu-2204:2024.01.2
     steps:
       - checkout
 
@@ -34,18 +34,18 @@ jobs:
       - run:
           name: Build containers and collect static
           command: |
-            docker-compose -f docker-compose.yml -f docker-compose.selenium.yml up -d
-            docker-compose -f docker-compose.yml -f docker-compose.selenium.yml exec django python manage.py collectstatic --noinput
+            docker compose -f docker-compose.yml -f docker-compose.selenium.yml up -d
+            docker compose -f docker-compose.yml -f docker-compose.selenium.yml exec django python manage.py collectstatic --noinput
 
       - run: docker-compose exec django flake8 src/
 
       - run:
           name: pytest
-          command: docker-compose -f docker-compose.yml -f docker-compose.selenium.yml exec django py.test src/ -m "not e2e"
+          command: docker compose -f docker-compose.yml -f docker-compose.selenium.yml exec django py.test src/ -m "not e2e"
 
       - run:
           name: e2e tests
-          command: docker-compose -f docker-compose.yml -f docker-compose.selenium.yml exec django py.test src/tests/functional/ -m e2e
+          command: docker compose -f docker-compose.yml -f docker-compose.selenium.yml exec django py.test src/tests/functional/ -m e2e
           no_output_timeout: 60m
 
       - store_artifacts:

.env_circleci

@@ -10,6 +10,7 @@ RABBITMQ_DEFAULT_USER=rabbit-username
 RABBITMQ_DEFAULT_PASS=rabbit-password-you-should-change
 RABBITMQ_PORT=5672
 RABBITMQ_HOST=rabbit
+WORKER_CONNECTION_TIMEOUT=100000000 # milliseconds
 
 FLOWER_BASIC_AUTH=root:password-you-should-change
 

.env_sample

@@ -23,6 +23,10 @@ RABBITMQ_DEFAULT_USER=rabbit-username
 RABBITMQ_DEFAULT_PASS=rabbit-password-you-should-change
 RABBITMQ_MANAGEMENT_PORT=15672
 RABBITMQ_PORT=5672
+WORKER_CONNECTION_TIMEOUT=100000000 # milliseconds
+#RABBITMQ_HTTP_PROXY=http://proxy-example:3128
+#RABBITMQ_HTTPS_PROXY=http://proxy-example:3128
+#RABBITMQ_NO_PROXY=localhost,172.0.0.0/8
 
 FLOWER_PUBLIC_PORT=5555
 
@@ -37,6 +41,9 @@ SELENIUM_HOSTNAME=selenium
 #EMAIL_HOST_PASSWORD=pass
 #EMAIL_PORT=587
 #EMAIL_USE_TLS=True
+#DEFAULT_FROM_EMAIL="Codabench <noreply@example.com>"
+#SERVER_EMAIL=noreply@example.com
+
 
 # -----------------------------------------------------------------------------
 # Storage
@@ -59,6 +66,22 @@ AWS_STORAGE_PRIVATE_BUCKET_NAME=private
 AWS_S3_ENDPOINT_URL=http://minio:9000/
 AWS_QUERYSTRING_AUTH=False
 
+
+# -----------------------------------------------------------------------------
+# Limit for re-running submission
+# This is used to limit users to rerun submissions
+# on default queue when number of submissions are < RERUN_SUBMISSION_LIMIT
+# -----------------------------------------------------------------------------
+RERUN_SUBMISSION_LIMIT=30
+
+
+# -----------------------------------------------------------------------------
+# Enable or disbale regular email sign-in an sign-up
+# -----------------------------------------------------------------------------
+ENABLE_SIGN_UP=True
+ENABLE_SIGN_IN=True
+
+
 # # S3 storage example
 # STORAGE_TYPE=s3
 # AWS_ACCESS_KEY_ID=12312312312312312331223

.github/CONTRIBUTING.md

@@ -0,0 +1,23 @@
+# HOW YOU CAN CONTRIBUTE TO THE CODABENCH PROJECT
+
+## 1. Being a Codabench user.
+
+- Create a user account on https://codalab.lisn.fr and on https://codabench.org.
+- Register on https://codabench.org to this existing competition (IRIS-tuto) https://www.codabench.org/competitions/1115/  and make a submission (from https://github.com/codalab/competition-examples/tree/master/codabench/iris): sample_result_submission and sample_code_submission. See https://github.com/codalab/codabench/wiki/User_Participating-in-a-Competition
+- Create your own private competition (from https://github.com/codalab/competition-examples/tree/master/codabench/ ). See https://github.com/codalab/codabench/wiki/Getting-started-with-Codabench
+
+ ## 2. Setting a local instance of Codabench.
+
+- Follow the tutorial in codabench wiki: https://github.com/codalab/codabench/wiki/Codabench-Installation. According to your hosting OS, you might have to tune your environment file a bit. Try without enabling the SSL protocol (doing so, you don't need a domain name for the server). Try using the embedded Minio storage solution instead of a private cloud storage.
+- If needed, you can also look into https://github.com/codalab/codabench/wiki/How-to-deploy-Codabench-on-your-server
+
+## 3. Using one's local instance
+
+- Create your own competition and play with it. You can look at the output logs of each different docker container.
+- Setting you as an admin of your platform (https://github.com/codalab/codabench/wiki/Administrator-procedures#give-superuser-privileges-to-an-user) and visit the Django Admin menu: https://github.com/codalab/codabench/wiki/Administrator-procedures#give-superuser-privileges-to-an-user
+
+## 4. Setting an autonomous computer-worker on your PC
+
+- Configure and launch the docker container: https://github.com/codalab/codabench/wiki/Compute-Worker-Management---Setup
+- Create a private queue on your new own competition on the production server codabench.org: https://github.com/codalab/codabench/wiki/Queue-Management#create-queue
+- Assign your own compute-worker to this private queue instead of the default queue.

.github/ISSUE_TEMPLATE.md

@@ -0,0 +1,19 @@
+***This is a template, please remove any non-relevant details to your issue***
+
+If you are competition participant:
+-----------------------------------
+
+- This Github repository is about the **platform** itself.
+- If you problem is specific to a competition, **please contact directly its organizers**.
+
+If you are an organizer and have problem hosting your competition:
+------------------------------------------------------------------
+
+- Please post a link to your competition.
+
+
+If you are having trouble using the site:
+-----------------------------------------
+
+- What browser and version are you using?
+- What is the URL of the problem? Codalab is an open source project, we may not be supporting the instance you are using!

.gitignore

@@ -13,6 +13,7 @@ var/
 var_*/
 certs/
 backups/
+logs/
 
 src/static/output.css
 src/static/output.js

Dockerfile.rabbitmq

@@ -0,0 +1,3 @@
+FROM rabbitmq:management
+ARG WORKER_CONNECTION_TIMEOUT
+RUN echo "consumer_timeout = $WORKER_CONNECTION_TIMEOUT" >> /etc/rabbitmq/conf.d/10-defaults.conf

compute_worker/compute_worker.py

@@ -26,9 +26,12 @@
 from kombu import Queue, Exchange
 from urllib3 import Retry
 
-
 logger = logging.getLogger()
 
+
+# -----------------------------------------------
+# Celery + Rabbit MQ
+# -----------------------------------------------
 # Init celery + rabbit queue definitions
 app = Celery()
 app.config_from_object('celery_config')  # grabs celery_config.py
@@ -38,13 +41,20 @@
 ]
 
 
+# -----------------------------------------------
+# Directories
+# -----------------------------------------------
 # Setup base directories used by all submissions
 # note: we need to pass this directory to docker-compose so it knows where to store things!
 HOST_DIRECTORY = os.environ.get("HOST_DIRECTORY", "/tmp/codabench/")
 BASE_DIR = "/codabench/"  # base directory inside the container
 CACHE_DIR = os.path.join(BASE_DIR, "cache")
 MAX_CACHE_DIR_SIZE_GB = float(os.environ.get('MAX_CACHE_DIR_SIZE_GB', 10))
 
+
+# -----------------------------------------------
+# Submission status
+# -----------------------------------------------
 # Status options for submissions
 STATUS_NONE = "None"
 STATUS_SUBMITTING = "Submitting"
@@ -65,6 +75,10 @@
     STATUS_FAILED,
 )
 
+
+# -----------------------------------------------
+# Container Engine
+# -----------------------------------------------
 # Setup the container engine that we are using
 if os.environ.get("CONTAINER_ENGINE_EXECUTABLE"):
     CONTAINER_ENGINE_EXECUTABLE = os.environ.get("CONTAINER_ENGINE_EXECUTABLE")
@@ -75,9 +89,18 @@
     CONTAINER_ENGINE_EXECUTABLE = "docker"
 
 
+# -----------------------------------------------
+# Exceptions
+# -----------------------------------------------
 class SubmissionException(Exception):
     pass
 
+class DockerImagePullException(Exception):
+    pass
+
+class ExecutionTimeLimitExceeded(Exception):
+    pass
+
 
 # -----------------------------------------------------------------------------
 # The main compute worker entrypoint, this is how a job is ran at the highest
@@ -94,6 +117,8 @@ def run_wrapper(run_args):
         if run.is_scoring:
             run.push_scores()
         run.push_output()
+    except DockerImagePullException as e:
+        run._update_status(STATUS_FAILED, str(e))
     except SubmissionException as e:
         run._update_status(STATUS_FAILED, str(e))
     except SoftTimeLimitExceeded:
@@ -160,14 +185,14 @@ def is_valid_zip(zip_path):
         return False
 
 
-class ExecutionTimeLimitExceeded(Exception):
-    pass
-
-
 def alarm_handler(signum, frame):
     raise ExecutionTimeLimitExceeded
 
 
+# -----------------------------------------------
+# Class Run
+# Respnosible for running a submission inside a docker container
+# -----------------------------------------------
 class Run:
     """A "Run" in Codalab is composed of some program, some data to work with, and some signed URLs to upload results
     to. There is also a secret key to do special commands for just this submission.
@@ -309,7 +334,7 @@ def _update_submission(self, data):
 
         logger.info(f"Updating submission @ {url} with data = {data}")
 
-        resp = self.requests_session.patch(url, data, timeout=15)
+        resp = self.requests_session.patch(url, data, timeout=150)
         if resp.status_code == 200:
             logger.info("Submission updated successfully!")
         else:
@@ -335,13 +360,69 @@ def _update_status(self, status, extra_information=None):
 
     def _get_container_image(self, image_name):
         logger.info("Running pull for image: {}".format(image_name))
+        retries, max_retries = (0, 3)
+        while retries < max_retries:
+            try:
+                cmd = [CONTAINER_ENGINE_EXECUTABLE, 'pull', image_name]
+                container_engine_pull = check_output(cmd)
+                logger.info("Pull complete for image: {0} with output of {1}".format(image_name, container_engine_pull))
+                break  # Break if the loop is successful
+            except CalledProcessError as pull_error:
+                retries += 1
+                if retries >= max_retries:
+                    error_message = f"Pull for image: {image_name} returned a non-zero exit code! Check if the docker image exists on docker hub. {pull_error}"
+                    logger.info(error_message)
+                    # Prepare data to be sent to submissions api
+                    docker_pull_fail_data = {
+                        "type": "Docker_Image_Pull_Fail",
+                        "error_message": error_message,
+                        "is_scoring": self.is_scoring
+                    }
+                    # Send data to be written to ingestion logs
+                    self._update_submission(docker_pull_fail_data)
+                    # Send error through web socket to the frontend
+                    asyncio.run(self._send_data_through_socket(error_message))
+                    raise DockerImagePullException(f"Pull for {image_name} failed!")
+                else:
+                    logger.info("Failed. Retrying in 5 seconds...")
+                    time.sleep(5)  # Wait 5 seconds before retrying
+
+    async def _send_data_through_socket(self, error_message):
+        """
+        This function gets an error messages and sends it through a web socket. This function is used for sending
+        - Docker image pull failure logs
+        - Execution time limit exceeded logs
+        """
+        logger.info(f"Connecting to {self.websocket_url} to send docker image pull error")
+
+        # connect to web socket
+        websocket = await websockets.connect(self.websocket_url)
+
+        # define websocket errors
+        websocket_errors = (socket.gaierror, websockets.WebSocketException, websockets.ConnectionClosedError, ConnectionRefusedError)
+
         try:
-            cmd = [CONTAINER_ENGINE_EXECUTABLE, 'pull', image_name]
-            container_engine_pull = check_output(cmd)
-            logger.info("Pull complete for image: {0} with output of {1}".format(image_name, container_engine_pull))
-        except CalledProcessError:
-            logger.info("Pull for image: {} returned a non-zero exit code!")
-            raise SubmissionException(f"Pull for {image_name} failed!")
+            # send message
+            await websocket.send(json.dumps({
+                "kind": "stderr",
+                "message": error_message
+            }))
+
+        except websocket_errors:
+            # handle websocket errors
+            logger.info(f"Error sending failed through websocket")
+            try:
+                await websocket.close()
+            except Exception as e:
+                logger.error(e)
+        else:
+            # no error in websocket message sending
+            logger.info(f"Error sent successfully through websocket")
+
+        logger.info(f"Disconnecting from websocket {self.websocket_url}")
+
+        # close websocket
+        await websocket.close()
 
     def _get_bundle(self, url, destination, cache=True):
         """Downloads zip from url and unzips into destination. If cache=True then url is hashed and checked
@@ -384,7 +465,7 @@ def _get_bundle(self, url, destination, cache=True):
                     raise  # Re-raise the last caught BadZipFile exception
                 else:
                     logger.info("Failed. Retrying in 60 seconds...")
-                    time.sleep(60) # Wait 60 seconds before retrying
+                    time.sleep(60)  # Wait 60 seconds before retrying
         # Return the zip file path for other uses, e.g. for creating a MD5 hash to identify it
         return bundle_file
 
@@ -426,12 +507,25 @@ async def _run_container_engine_cmd(self, engine_cmd, kind):
         websocket = await websockets.connect(self.websocket_url)
         websocket_errors = (socket.gaierror, websockets.WebSocketException, websockets.ConnectionClosedError, ConnectionRefusedError)
 
+        # Function to read a line, if the line is larger than the buffer size we will
+        # return the buffer so we can continue reading until we get a newline, rather
+        # than getting a LimitOverrunError
+        async def _readline_or_chunk(stream):
+            try:
+                return await stream.readuntil(b"\n")
+            except asyncio.exceptions.IncompleteReadError as e:
+                # Just return what has been read so far
+                return e.partial
+            except asyncio.exceptions.LimitOverrunError as e:
+                # If we get a LimitOverrunError, we will return the buffer so we can continue reading
+                return await stream.read(e.consumed)
+
         while any(v["continue"] for k, v in self.logs[kind].items() if k in ['stdout', 'stderr']):
             try:
                 logs = [self.logs[kind][key] for key in ('stdout', 'stderr')]
                 for value in logs:
                     try:
-                        out = await asyncio.wait_for(value["stream"].readline(), timeout=.1)
+                        out = await asyncio.wait_for(_readline_or_chunk(value["stream"]), timeout=.1)
                         if out:
                             value["data"] += out
                             print("WS: " + str(out))
@@ -610,7 +704,7 @@ def _put_dir(self, url, directory):
             start_time = time.time()
             zip_path = make_archive(os.path.join(self.root_dir, str(uuid.uuid4())), 'zip', directory)
             duration = time.time() - start_time
-            logger.info("Time needed to zip archive: {duration} seconds.")
+            logger.info(f"Time needed to zip archive: {duration} seconds.")
             if is_valid_zip(zip_path): # Check zip integrity
                 self._put_file(url, file=zip_path) # Send the file
                 break # Leave the loop in case of success
@@ -627,7 +721,7 @@ def _put_file(self, url, file=None, raw_data=None, content_type='application/zip
         """
         if file and raw_data:
             raise Exception("Cannot put both a file and raw_data")
-        
+
         headers = {
             # For Azure only, other systems ignore these headers
             'x-ms-blob-type': 'BlockBlob',
@@ -731,7 +825,19 @@ def start(self):
         try:
             loop.run_until_complete(gathered_tasks)
         except ExecutionTimeLimitExceeded:
-            raise SubmissionException(f"Execution Time Limit exceeded. Limit was {self.execution_time_limit} seconds")
+            error_message = f"Execution Time Limit exceeded. Limit was {self.execution_time_limit} seconds"
+            logger.info(error_message)
+            # Prepare data to be sent to submissions api
+            execution_time_limit_exceeded_data = {
+                "type": "Execution_Time_Limit_Exceeded",
+                "error_message": error_message,
+                "is_scoring": self.is_scoring
+            }
+            # Send data to be written to ingestion/scoring std_err
+            self._update_submission(execution_time_limit_exceeded_data)
+            # Send error through web socket to the frontend
+            asyncio.run(self._send_data_through_socket(error_message))
+            raise SubmissionException(error_message)
         finally:
             self.watch = False
             for kind, logs in self.logs.items():