Add handler for deleting cache after interrupted import

.vscode/launch.json

@@ -270,6 +270,28 @@
             "env": {},
             "console": "internalConsole"
         },
+        {
+            "name": "server: RQ - cleaning",
+            "type": "python",
+            "request": "launch",
+            "stopOnEntry": false,
+            "justMyCode": false,
+            "python": "${command:python.interpreterPath}",
+            "program": "${workspaceRoot}/manage.py",
+            "args": [
+                "rqworker",
+                "cleaning",
+                "--worker-class",
+                "cvat.rqworker.SimpleWorker"
+            ],
+            "django": true,
+            "cwd": "${workspaceFolder}",
+            "env": {
+                "DJANGO_LOG_SERVER_HOST": "localhost",
+                "DJANGO_LOG_SERVER_PORT": "8282"
+            },
+            "console": "internalConsole"
+        },
         {
             "name": "server: migrate",
             "type": "python",
@@ -433,6 +455,7 @@
                 "server: RQ - annotation",
                 "server: RQ - webhooks",
                 "server: RQ - scheduler",
+                "server: RQ - cleaning",
                 "server: git",
             ]
         }

cvat/apps/engine/handlers.py

@@ -0,0 +1,22 @@
+# Copyright (C) 2023 CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
+from pathlib import Path
+from time import time
+from django.conf import settings
+from cvat.apps.engine.log import slogger
+
+
+def clear_import_cache(path: Path, creation_time: float) -> None:
+    """
+    This function checks and removes the import files if they have not been removed from rq import jobs.
+    This means that for some reason file was uploaded to CVAT server but rq import job was not created.
+
+    Args:
+        path (Path): path to file
+        creation_time (float): file creation time
+    """
+    if path.is_file() and (time() - creation_time + 1) >= settings.RUN_CLEAN_IMPORT_CACHE_FUNC_AFTER.total_seconds():
+        path.unlink()
+        slogger.glob.warning(f"The file {str(path)} was removed from cleaning job.")

cvat/apps/engine/mixins.py

@@ -7,19 +7,22 @@
 import json
 import os
 import uuid
-import django_rq
 from dataclasses import asdict, dataclass
 from distutils.util import strtobool
+from pathlib import Path
+from tempfile import NamedTemporaryFile
 from unittest import mock
 
+import django_rq
 from django.conf import settings
 from rest_framework import mixins, status
 from rest_framework.response import Response
-from tempfile import NamedTemporaryFile
 
 from cvat.apps.engine.location import StorageType, get_location_configuration
+from cvat.apps.engine.log import slogger
 from cvat.apps.engine.models import Location
 from cvat.apps.engine.serializers import DataSerializer
+from cvat.apps.engine.handlers import clear_import_cache
 from cvat.apps.engine.utils import get_import_rq_id
 
 
@@ -229,7 +232,7 @@ def init_tus_upload(self, request):
                 # we need to create unique temp file here because
                 # users can try to import backups with the same name at the same time
                 with NamedTemporaryFile(prefix=f'cvat-backup-{filename}-by-{request.user}', suffix='.zip', dir=self.get_upload_dir()) as tmp_file:
-                    filename = tmp_file.name
+                    filename = os.path.relpath(tmp_file.name, self.get_upload_dir())
                 metadata['filename'] = filename
             file_path = os.path.join(self.get_upload_dir(), filename)
             file_exists = os.path.lexists(file_path) and import_type != 'backup'
@@ -260,6 +263,21 @@ def init_tus_upload(self, request):
             location = request.build_absolute_uri()
             if 'HTTP_X_FORWARDED_HOST' not in request.META:
                 location = request.META.get('HTTP_ORIGIN') + request.META.get('PATH_INFO')
+
+            if import_type in ('backup', 'annotations', 'datasets'):
+                scheduler = django_rq.get_scheduler(settings.CVAT_QUEUES.CLEANING.value)
+                path = Path(self.get_upload_dir()) / tus_file.filename
+                cleaning_job = scheduler.enqueue_in(time_delta=settings.RUN_CLEAN_IMPORT_CACHE_FUNC_AFTER,
+                    func=clear_import_cache,
+                    path=path,
+                    creation_time=Path(tus_file.file_path).stat().st_atime
+                )
+                slogger.glob.info(
+                    f'The cleaning job {cleaning_job.id} is queued.'
+                    f'The check that the file {path} is deleted will be carried out after '
+                    f'{settings.RUN_CLEAN_IMPORT_CACHE_FUNC_AFTER}.'
+                )
+
             return self._tus_response(
                 status=status.HTTP_201_CREATED,
                 extra_headers={'Location': '{}{}'.format(location, tus_file.file_id),

cvat/apps/engine/utils.py

@@ -234,7 +234,12 @@ def get_list_view_name(model):
     }
 
 
-def get_import_rq_id(resource_type: str, resource_id: int, subresource_type: str, user: str) -> str:
+def get_import_rq_id(
+    resource_type: str,
+    resource_id: int,
+    subresource_type: str,
+    user: str,
+) -> str:
     # import:<task|project|job>-<id|uuid>-<annotations|dataset|backup>-by-<user>
     return f"import:{resource_type}-{resource_id}-{subresource_type}-by-{user}"
 

cvat/apps/engine/views.py

@@ -2226,9 +2226,6 @@ def _download_file_from_bucket(db_storage, filename, key):
 def _import_annotations(request, rq_id_template, rq_func, db_obj, format_name,
                         filename=None, location_conf=None, conv_mask_to_poly=True):
 
-    def _is_there_past_non_deleted_job():
-        return rq_job and request.method == 'POST' and (rq_job.is_finished or rq_job.is_failed)
-
     format_desc = {f.DISPLAY_NAME: f
         for f in dm.views.get_import_formats()}.get(format_name)
     if format_desc is None:
@@ -2248,9 +2245,13 @@ def _is_there_past_non_deleted_job():
     if rq_id_should_be_checked and rq_id_template.format(db_obj.pk, request.user) != rq_id:
         return Response(status=status.HTTP_403_FORBIDDEN)
 
-    if _is_there_past_non_deleted_job():
-        rq_job.delete()
-        rq_job = queue.fetch_job(rq_id)
+    if rq_job and request.method == 'POST':
+        # If there is a previous job that has not been deleted
+        if rq_job.is_finished or rq_job.is_failed:
+            rq_job.delete()
+            rq_job = queue.fetch_job(rq_id)
+        else:
+            return Response(status=status.HTTP_409_CONFLICT, data='Import job already exists')
 
     if not rq_job:
         # If filename is specified we consider that file was uploaded via TUS, so it exists in filesystem

cvat/settings/base.py

@@ -295,6 +295,7 @@ class CVAT_QUEUES(Enum):
     AUTO_ANNOTATION = 'annotation'
     WEBHOOKS = 'webhooks'
     NOTIFICATIONS = 'notifications'
+    CLEANING = 'cleaning'
 
 RQ_QUEUES = {
     CVAT_QUEUES.IMPORT_DATA.value: {
@@ -327,6 +328,12 @@ class CVAT_QUEUES(Enum):
         'DB': 0,
         'DEFAULT_TIMEOUT': '1h'
     },
+    CVAT_QUEUES.CLEANING.value: {
+        'HOST': 'localhost',
+        'PORT': 6379,
+        'DB': 0,
+        'DEFAULT_TIMEOUT': '1h'
+    },
 }
 
 NUCLIO = {
@@ -663,3 +670,4 @@ class CVAT_QUEUES(Enum):
 
 IMPORT_CACHE_FAILED_TTL = timedelta(days=90).total_seconds()
 IMPORT_CACHE_SUCCESS_TTL = timedelta(hours=1).total_seconds()
+RUN_CLEAN_IMPORT_CACHE_FUNC_AFTER = timedelta(hours=2)

cvat/settings/rest-api-testing.py

@@ -0,0 +1,7 @@
+# Copyright (C) 2023 CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
+from .production import *
+
+RUN_CLEAN_IMPORT_CACHE_FUNC_AFTER = timedelta(seconds=10)

docker-compose.yml

@@ -178,6 +178,28 @@ services:
     networks:
       - cvat
 
+  cvat_worker_cleaning:
+    container_name: cvat_worker_cleaning
+    image: cvat/server:${CVAT_VERSION:-dev}
+    restart: always
+    depends_on:
+      - cvat_redis
+      - cvat_db
+    environment:
+      CVAT_REDIS_HOST: 'cvat_redis'
+      CVAT_POSTGRES_HOST: 'cvat_db'
+      DJANGO_LOG_SERVER_HOST: vector
+      DJANGO_LOG_SERVER_PORT: 80
+      no_proxy: clickhouse,grafana,vector,nuclio,opa,${no_proxy:-}
+      NUMPROCS: 1
+    command: -c supervisord/worker.cleaning.conf
+    volumes:
+      - cvat_data:/home/django/data
+      - cvat_keys:/home/django/keys
+      - cvat_logs:/home/django/logs
+    networks:
+      - cvat
+
   cvat_ui:
     container_name: cvat_ui
     image: cvat/ui:${CVAT_VERSION:-dev}

supervisord/worker.cleaning.conf

@@ -0,0 +1,33 @@
+[unix_http_server]
+file = /tmp/supervisord/supervisor.sock
+
+[supervisorctl]
+serverurl = unix:///tmp/supervisord/supervisor.sock
+
+
+[rpcinterface:supervisor]
+supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface
+
+[supervisord]
+nodaemon=true
+logfile=%(ENV_HOME)s/logs/supervisord.log ; supervisord log file
+logfile_maxbytes=50MB       ; maximum size of logfile before rotation
+logfile_backups=10          ; number of backed up logfiles
+loglevel=debug              ; info, debug, warn, trace
+pidfile=/tmp/supervisord/supervisord.pid ; pidfile location
+childlogdir=%(ENV_HOME)s/logs/            ; where child log files will live
+
+[program:ssh-agent]
+command=bash -c "rm /tmp/ssh-agent.sock -f && /usr/bin/ssh-agent -d -a /tmp/ssh-agent.sock"
+priority=1
+autorestart=true
+
+[program:rqworker_cleaning]
+command=%(ENV_HOME)s/wait-for-it.sh %(ENV_CVAT_REDIS_HOST)s:6379 -t 0 -- bash -ic " \
+    exec python3 %(ENV_HOME)s/manage.py rqworker -v 3 cleaning \
+        --worker-class cvat.rqworker.DefaultWorker \
+    "
+environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock",VECTOR_EVENT_HANDLER="SynchronousLogstashHandler"
+numprocs=%(ENV_NUMPROCS)s
+process_name=rqworker_cleaning_%(process_num)s
+

tests/python/rest_api/test_tasks.py

@@ -1523,6 +1523,7 @@ def test_can_report_correct_completed_jobs_count(tasks, jobs, admin_user):
         assert task.jobs.completed == 1
 
 
+@pytest.mark.usefixtures("restore_cvat_data")
 class TestImportTaskAnnotations:
     def _make_client(self) -> Client:
         return Client(BASE_URL, config=Config(status_check_period=0.01))
@@ -1549,8 +1550,8 @@ def _delete_annotations(self, task_id):
             (_, response) = api_client.tasks_api.destroy_annotations(id=task_id)
             assert response.status == HTTPStatus.NO_CONTENT
 
-    @pytest.mark.parametrize("task_id", [20])
-    def test_can_import_annotations_after_previous_unclear_import(self, task_id):
+    def test_can_import_annotations_after_previous_unclear_import(self, tasks_with_shapes):
+        task_id = tasks_with_shapes[0]["id"]
         self._check_annotations(task_id)
 
         filename = self.tmp_dir / f"task_{task_id}_coco.zip"
@@ -1582,3 +1583,31 @@ def test_can_import_annotations_after_previous_unclear_import(self, task_id):
         self._delete_annotations(task_id)
         task.import_annotations(self.format, filename)
         self._check_annotations(task_id)
+
+    @pytest.mark.timeout(40)
+    def test_can_import_annotations_after_previous_interrupted_upload(self, tasks_with_shapes):
+        task_id = tasks_with_shapes[0]["id"]
+        self._check_annotations(task_id)
+
+        filename = self.tmp_dir / f"task_{task_id}_coco.zip"
+        task = self.client.tasks.retrieve(task_id)
+        task.export_dataset(self.format, filename, include_images=False)
+        self._delete_annotations(task_id)
+
+        params = {"format": self.format, "filename": filename.name}
+        url = self.client.api_map.make_endpoint_url(
+            self.client.api_client.tasks_api.create_annotations_endpoint.path
+        ).format(id=task_id)
+
+        uploader = Uploader(self.client)
+        uploader._tus_start_upload(url, query_params=params)
+        uploader._upload_file_data_with_tus(
+            url, filename, meta=params, logger=self.client.logger.debug
+        )
+        sleep(30)
+        assert not int(
+            subprocess.check_output(
+                f'docker exec -it test_cvat_server_1 bash -c "ls data/tasks/{task_id}/tmp | wc -l"',
+                shell=True,
+            )
+        )

tests/python/shared/fixtures/init.py

@@ -22,14 +22,14 @@
 PREFIX = "test"
 
 CONTAINER_NAME_FILES = ["docker-compose.tests.yml"]
-
+OVERRIDE_FILES = ['tests/docker-compose.override.yml']
 
 DC_FILES = [
     "docker-compose.dev.yml",
     "tests/docker-compose.file_share.yml",
     "tests/docker-compose.minio.yml",
     "tests/docker-compose.test_servers.yml",
-] + CONTAINER_NAME_FILES
+] + CONTAINER_NAME_FILES + OVERRIDE_FILES
 
 
 def pytest_addoption(parser):