Splitted default queue to import/export, fixed django-rq admin page (#5555)

main
Andrey Zhavoronkov 3 years ago committed by GitHub
parent 0ea14d23a8
commit fd7d8024f9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -133,7 +133,7 @@
} }
}, },
{ {
"name": "server: RQ - default", "name": "server: RQ - import",
"type": "python", "type": "python",
"request": "launch", "request": "launch",
"stopOnEntry": false, "stopOnEntry": false,
@ -142,7 +142,26 @@
"program": "${workspaceRoot}/manage.py", "program": "${workspaceRoot}/manage.py",
"args": [ "args": [
"rqworker", "rqworker",
"default", "import",
"--worker-class",
"cvat.rqworker.SimpleWorker",
],
"django": true,
"cwd": "${workspaceFolder}",
"env": {},
"console": "internalConsole"
},
{
"name": "server: RQ - export",
"type": "python",
"request": "launch",
"stopOnEntry": false,
"justMyCode": false,
"python": "${command:python.interpreterPath}",
"program": "${workspaceRoot}/manage.py",
"args": [
"rqworker",
"export",
"--worker-class", "--worker-class",
"cvat.rqworker.SimpleWorker", "cvat.rqworker.SimpleWorker",
], ],
@ -161,6 +180,8 @@
"program": "${workspaceRoot}/manage.py", "program": "${workspaceRoot}/manage.py",
"args": [ "args": [
"rqscheduler", "rqscheduler",
"--queue",
"export"
], ],
"django": true, "django": true,
"cwd": "${workspaceFolder}", "cwd": "${workspaceFolder}",
@ -168,7 +189,7 @@
"console": "internalConsole" "console": "internalConsole"
}, },
{ {
"name": "server: RQ - low", "name": "server: RQ - annotation",
"type": "python", "type": "python",
"request": "launch", "request": "launch",
"justMyCode": false, "justMyCode": false,
@ -177,7 +198,7 @@
"program": "${workspaceRoot}/manage.py", "program": "${workspaceRoot}/manage.py",
"args": [ "args": [
"rqworker", "rqworker",
"low", "annotation",
"--worker-class", "--worker-class",
"cvat.rqworker.SimpleWorker", "cvat.rqworker.SimpleWorker",
], ],
@ -379,8 +400,9 @@
"name": "server: debug", "name": "server: debug",
"configurations": [ "configurations": [
"server: django", "server: django",
"server: RQ - default", "server: RQ - import",
"server: RQ - low", "server: RQ - export",
"server: RQ - annotation",
"server: RQ - webhooks", "server: RQ - webhooks",
"server: RQ - scheduler", "server: RQ - scheduler",
"server: git", "server: git",

@ -28,6 +28,6 @@ services:
extra_hosts: extra_hosts:
- "host.docker.internal:host-gateway" - "host.docker.internal:host-gateway"
cvat_worker_low: cvat_worker_annotation:
extra_hosts: extra_hosts:
- "host.docker.internal:host-gateway" - "host.docker.internal:host-gateway"

@ -11,6 +11,7 @@ import django_rq
from datumaro.util.os_util import make_file_name from datumaro.util.os_util import make_file_name
from datumaro.util import to_snake_case from datumaro.util import to_snake_case
from django.utils import timezone from django.utils import timezone
from django.conf import settings
import cvat.apps.dataset_manager.task as task import cvat.apps.dataset_manager.task as task
import cvat.apps.dataset_manager.project as project import cvat.apps.dataset_manager.project as project
@ -83,7 +84,7 @@ def export(dst_format, project_id=None, task_id=None, job_id=None, server_url=No
os.replace(temp_file, output_path) os.replace(temp_file, output_path)
archive_ctime = osp.getctime(output_path) archive_ctime = osp.getctime(output_path)
scheduler = django_rq.get_scheduler() scheduler = django_rq.get_scheduler(settings.CVAT_QUEUES.EXPORT_DATA.value)
cleaning_job = scheduler.enqueue_in(time_delta=cache_ttl, cleaning_job = scheduler.enqueue_in(time_delta=cache_ttl,
func=clear_export_cache, func=clear_export_cache,
file_path=output_path, file_path=output_path,

@ -15,6 +15,7 @@ import django_rq
import git import git
from django.db import transaction from django.db import transaction
from django.utils import timezone from django.utils import timezone
from django.conf import settings
from cvat.apps.dataset_manager.formats.registry import format_for from cvat.apps.dataset_manager.formats.registry import format_for
from cvat.apps.dataset_manager.task import export_task from cvat.apps.dataset_manager.task import export_task
@ -427,7 +428,7 @@ def get(tid, user):
response['url']['value'] = '{} [{}]'.format(db_git.url, db_git.path) response['url']['value'] = '{} [{}]'.format(db_git.url, db_git.path)
try: try:
rq_id = "git.push.{}".format(tid) rq_id = "git.push.{}".format(tid)
queue = django_rq.get_queue('default') queue = django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value)
rq_job = queue.fetch_job(rq_id) rq_job = queue.fetch_job(rq_id)
if rq_job is not None and (rq_job.is_queued or rq_job.is_started): if rq_job is not None and (rq_job.is_queued or rq_job.is_started):
db_git.status = GitStatusChoice.SYNCING db_git.status = GitStatusChoice.SYNCING

@ -5,6 +5,7 @@ import http.client
from django.http import HttpResponseBadRequest, HttpResponse from django.http import HttpResponseBadRequest, HttpResponse
from rules.contrib.views import permission_required, objectgetter from rules.contrib.views import permission_required, objectgetter
from django.conf import settings
from rest_framework.permissions import IsAuthenticated from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response from rest_framework.response import Response
@ -37,7 +38,7 @@ def _legacy_api_view(allowed_method_names=None):
@_legacy_api_view() @_legacy_api_view()
def check_process(request, rq_id): def check_process(request, rq_id):
try: try:
queue = django_rq.get_queue('default') queue = django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value)
rq_job = queue.fetch_job(rq_id) rq_job = queue.fetch_job(rq_id)
if rq_job is not None: if rq_job is not None:
@ -65,7 +66,7 @@ def create(request: Request, tid):
export_format = body.get("format") export_format = body.get("format")
lfs = body["lfs"] lfs = body["lfs"]
rq_id = "git.create.{}".format(tid) rq_id = "git.create.{}".format(tid)
queue = django_rq.get_queue("default") queue = django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value)
queue.enqueue_call(func = CVATGit.initial_create, args = (tid, path, export_format, lfs, request.user), job_id = rq_id) queue.enqueue_call(func = CVATGit.initial_create, args = (tid, path, export_format, lfs, request.user), job_id = rq_id)
return Response({ "rq_id": rq_id }) return Response({ "rq_id": rq_id })
@ -80,7 +81,7 @@ def push_repository(request: Request, tid):
slogger.task[tid].info("push repository request") slogger.task[tid].info("push repository request")
rq_id = "git.push.{}".format(tid) rq_id = "git.push.{}".format(tid)
queue = django_rq.get_queue('default') queue = django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value)
queue.enqueue_call(func = CVATGit.push, args = (tid, request.user, request.scheme, request.get_host()), job_id = rq_id) queue.enqueue_call(func = CVATGit.push, args = (tid, request.user, request.scheme, request.get_host()), job_id = rq_id)
return Response({ "rq_id": rq_id }) return Response({ "rq_id": rq_id })

@ -712,7 +712,7 @@ def _create_backup(db_instance, Exporter, output_path, logger, cache_ttl):
os.replace(temp_file, output_path) os.replace(temp_file, output_path)
archive_ctime = os.path.getctime(output_path) archive_ctime = os.path.getctime(output_path)
scheduler = django_rq.get_scheduler() scheduler = django_rq.get_scheduler(settings.CVAT_QUEUES.IMPORT_DATA.value)
cleaning_job = scheduler.enqueue_in(time_delta=cache_ttl, cleaning_job = scheduler.enqueue_in(time_delta=cache_ttl,
func=clear_export_cache, func=clear_export_cache,
file_path=output_path, file_path=output_path,
@ -731,7 +731,7 @@ def _create_backup(db_instance, Exporter, output_path, logger, cache_ttl):
log_exception(logger) log_exception(logger)
raise raise
def export(db_instance, request): def export(db_instance, request, queue_name):
action = request.query_params.get('action', None) action = request.query_params.get('action', None)
filename = request.query_params.get('filename', None) filename = request.query_params.get('filename', None)
@ -740,13 +740,13 @@ def export(db_instance, request):
"Unexpected action specified for the request") "Unexpected action specified for the request")
if isinstance(db_instance, Task): if isinstance(db_instance, Task):
filename_prefix = 'task' obj_type = 'task'
logger = slogger.task[db_instance.pk] logger = slogger.task[db_instance.pk]
Exporter = TaskExporter Exporter = TaskExporter
cache_ttl = TASK_CACHE_TTL cache_ttl = TASK_CACHE_TTL
use_target_storage_conf = request.query_params.get('use_default_location', True) use_target_storage_conf = request.query_params.get('use_default_location', True)
elif isinstance(db_instance, Project): elif isinstance(db_instance, Project):
filename_prefix = 'project' obj_type = 'project'
logger = slogger.project[db_instance.pk] logger = slogger.project[db_instance.pk]
Exporter = ProjectExporter Exporter = ProjectExporter
cache_ttl = PROJECT_CACHE_TTL cache_ttl = PROJECT_CACHE_TTL
@ -762,8 +762,8 @@ def export(db_instance, request):
field_name=StorageType.TARGET field_name=StorageType.TARGET
) )
queue = django_rq.get_queue("default") queue = django_rq.get_queue(queue_name)
rq_id = "/api/{}s/{}/backup".format(filename_prefix, db_instance.pk) rq_id = f"export:{obj_type}.id{db_instance.pk}-by-{request.user}"
rq_job = queue.fetch_job(rq_id) rq_job = queue.fetch_job(rq_id)
if rq_job: if rq_job:
last_project_update_time = timezone.localtime(db_instance.updated_date) last_project_update_time = timezone.localtime(db_instance.updated_date)
@ -780,7 +780,7 @@ def export(db_instance, request):
timestamp = datetime.strftime(last_project_update_time, timestamp = datetime.strftime(last_project_update_time,
"%Y_%m_%d_%H_%M_%S") "%Y_%m_%d_%H_%M_%S")
filename = filename or "{}_{}_backup_{}{}".format( filename = filename or "{}_{}_backup_{}{}".format(
filename_prefix, db_instance.name, timestamp, obj_type, db_instance.name, timestamp,
os.path.splitext(file_path)[1]).lower() os.path.splitext(file_path)[1]).lower()
location = location_conf.get('location') location = location_conf.get('location')
@ -820,7 +820,7 @@ def export(db_instance, request):
ttl = dm.views.PROJECT_CACHE_TTL.total_seconds() ttl = dm.views.PROJECT_CACHE_TTL.total_seconds()
queue.enqueue_call( queue.enqueue_call(
func=_create_backup, func=_create_backup,
args=(db_instance, Exporter, '{}_backup.zip'.format(filename_prefix), logger, cache_ttl), args=(db_instance, Exporter, '{}_backup.zip'.format(obj_type), logger, cache_ttl),
job_id=rq_id, job_id=rq_id,
meta={ 'request_time': timezone.localtime() }, meta={ 'request_time': timezone.localtime() },
result_ttl=ttl, failure_ttl=ttl) result_ttl=ttl, failure_ttl=ttl)
@ -834,8 +834,7 @@ def _download_file_from_bucket(db_storage, filename, key):
with open(filename, 'wb+') as f: with open(filename, 'wb+') as f:
f.write(data.getbuffer()) f.write(data.getbuffer())
def _import(importer, request, rq_id, Serializer, file_field_name, location_conf, filename=None): def _import(importer, request, queue, rq_id, Serializer, file_field_name, location_conf, filename=None):
queue = django_rq.get_queue("default")
rq_job = queue.fetch_job(rq_id) rq_job = queue.fetch_job(rq_id)
if not rq_job: if not rq_job:
@ -905,11 +904,11 @@ def _import(importer, request, rq_id, Serializer, file_field_name, location_conf
def get_backup_dirname(): def get_backup_dirname():
return settings.TMP_FILES_ROOT return settings.TMP_FILES_ROOT
def import_project(request, filename=None): def import_project(request, queue_name, filename=None):
if 'rq_id' in request.data: if 'rq_id' in request.data:
rq_id = request.data['rq_id'] rq_id = request.data['rq_id']
else: else:
rq_id = "{}@/api/projects/{}/import".format(request.user, uuid.uuid4()) rq_id = f"import:project.{uuid.uuid4()}-by-{request.user}"
Serializer = ProjectFileSerializer Serializer = ProjectFileSerializer
file_field_name = 'project_file' file_field_name = 'project_file'
@ -918,9 +917,12 @@ def import_project(request, filename=None):
field_name=StorageType.SOURCE, field_name=StorageType.SOURCE,
) )
queue = django_rq.get_queue(queue_name)
return _import( return _import(
importer=_import_project, importer=_import_project,
request=request, request=request,
queue=queue,
rq_id=rq_id, rq_id=rq_id,
Serializer=Serializer, Serializer=Serializer,
file_field_name=file_field_name, file_field_name=file_field_name,
@ -928,11 +930,11 @@ def import_project(request, filename=None):
filename=filename filename=filename
) )
def import_task(request, filename=None): def import_task(request, queue_name, filename=None):
if 'rq_id' in request.data: if 'rq_id' in request.data:
rq_id = request.data['rq_id'] rq_id = request.data['rq_id']
else: else:
rq_id = "{}@/api/tasks/{}/import".format(request.user, uuid.uuid4()) rq_id = f"import:task.{uuid.uuid4()}-by-{request.user}"
Serializer = TaskFileSerializer Serializer = TaskFileSerializer
file_field_name = 'task_file' file_field_name = 'task_file'
@ -941,9 +943,12 @@ def import_task(request, filename=None):
field_name=StorageType.SOURCE field_name=StorageType.SOURCE
) )
queue = django_rq.get_queue(queue_name)
return _import( return _import(
importer=_import_task, importer=_import_task,
request=request, request=request,
queue=queue,
rq_id=rq_id, rq_id=rq_id,
Serializer=Serializer, Serializer=Serializer,
file_field_name=file_field_name, file_field_name=file_field_name,

@ -246,7 +246,7 @@ class UploadMixin:
class AnnotationMixin: class AnnotationMixin:
def export_annotations(self, request, pk, db_obj, export_func, callback, get_data=None): def export_annotations(self, request, pk, db_obj, export_func, callback, get_data=None):
format_name = request.query_params.get("format") format_name = request.query_params.get("format", "")
action = request.query_params.get("action", "").lower() action = request.query_params.get("action", "").lower()
filename = request.query_params.get("filename", "") filename = request.query_params.get("filename", "")
@ -259,7 +259,8 @@ class AnnotationMixin:
field_name=StorageType.TARGET, field_name=StorageType.TARGET,
) )
rq_id = "/api/{}/{}/annotations/{}".format(self._object.__class__.__name__.lower(), pk, format_name) object_name = self._object.__class__.__name__.lower()
rq_id = f"export:annotations-for-{object_name}.id{pk}-in-{format_name.replace(' ', '_')}-format"
if format_name: if format_name:
return export_func(db_instance=self._object, return export_func(db_instance=self._object,
@ -316,13 +317,21 @@ class AnnotationMixin:
class SerializeMixin: class SerializeMixin:
def serialize(self, request, export_func): def serialize(self, request, export_func):
db_object = self.get_object() # force to call check_object_permissions db_object = self.get_object() # force to call check_object_permissions
return export_func(db_object, request) return export_func(
db_object,
request,
queue_name=settings.CVAT_QUEUES.EXPORT_DATA.value,
)
def deserialize(self, request, import_func): def deserialize(self, request, import_func):
location = request.query_params.get("location", Location.LOCAL) location = request.query_params.get("location", Location.LOCAL)
if location == Location.CLOUD_STORAGE: if location == Location.CLOUD_STORAGE:
file_name = request.query_params.get("filename", "") file_name = request.query_params.get("filename", "")
return import_func(request, filename=file_name) return import_func(
request,
queue_name=settings.CVAT_QUEUES.IMPORT_DATA.value,
filename=file_name,
)
return self.upload_data(request) return self.upload_data(request)

@ -38,11 +38,11 @@ from .cloud_provider import db_storage_to_storage_instance
############################# Low Level server API ############################# Low Level server API
def create(tid, data): def create(tid, data, username):
"""Schedule the task""" """Schedule the task"""
q = django_rq.get_queue('default') q = django_rq.get_queue(settings.CVAT_QUEUES.IMPORT_DATA.value)
q.enqueue_call(func=_create_thread, args=(tid, data), q.enqueue_call(func=_create_thread, args=(tid, data),
job_id="/api/tasks/{}".format(tid)) job_id=f"create:task.id{tid}-by-{username}")
@transaction.atomic @transaction.atomic
def rq_handler(job, exc_type, exc_value, traceback): def rq_handler(job, exc_type, exc_value, traceback):

@ -407,7 +407,7 @@ class ProjectViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
resource_type_field_name=None resource_type_field_name=None
), ),
responses={ responses={
'202': OpenApiResponse(description='Exporting has been started'), '202': OpenApiResponse(description='Importing has been started'),
'400': OpenApiResponse(description='Failed to import dataset'), '400': OpenApiResponse(description='Failed to import dataset'),
'405': OpenApiResponse(description='Format is not available'), '405': OpenApiResponse(description='Format is not available'),
}) })
@ -415,6 +415,7 @@ class ProjectViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
url_path=r'dataset/?$') url_path=r'dataset/?$')
def dataset(self, request, pk): def dataset(self, request, pk):
self._object = self.get_object() # force to call check_object_permissions self._object = self.get_object() # force to call check_object_permissions
rq_id = f"import:dataset-for-porject.id{pk}-by-{request.user}"
if request.method in {'POST', 'OPTIONS'}: if request.method in {'POST', 'OPTIONS'}:
return self.import_annotations( return self.import_annotations(
@ -423,13 +424,13 @@ class ProjectViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
db_obj=self._object, db_obj=self._object,
import_func=_import_project_dataset, import_func=_import_project_dataset,
rq_func=dm.project.import_dataset_as_project, rq_func=dm.project.import_dataset_as_project,
rq_id=f"/api/project/{pk}/dataset_import", rq_id=rq_id,
) )
else: else:
action = request.query_params.get("action", "").lower() action = request.query_params.get("action", "").lower()
if action in ("import_status",): if action in ("import_status",):
queue = django_rq.get_queue("default") queue = django_rq.get_queue(settings.CVAT_QUEUES.IMPORT_DATA.value)
rq_job = queue.fetch_job(f"/api/project/{pk}/dataset_import") rq_job = queue.fetch_job(rq_id)
if rq_job is None: if rq_job is None:
return Response(status=status.HTTP_404_NOT_FOUND) return Response(status=status.HTTP_404_NOT_FOUND)
elif rq_job.is_finished: elif rq_job.is_finished:
@ -449,7 +450,10 @@ class ProjectViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
) )
else: else:
return Response( return Response(
data=self._get_rq_response('default', f'/api/project/{pk}/dataset_import'), data=self._get_rq_response(
settings.CVAT_QUEUES.IMPORT_DATA.value,
rq_id,
),
status=status.HTTP_202_ACCEPTED status=status.HTTP_202_ACCEPTED
) )
else: else:
@ -495,7 +499,7 @@ class ProjectViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
return _import_project_dataset( return _import_project_dataset(
request=request, request=request,
filename=uploaded_file, filename=uploaded_file,
rq_id=f"/api/project/{self._object.pk}/dataset_import", rq_id=f"import:dataset-for-porject.id{self._object.pk}-by-{request.user}",
rq_func=dm.project.import_dataset_as_project, rq_func=dm.project.import_dataset_as_project,
pk=self._object.pk, pk=self._object.pk,
format_name=format_name, format_name=format_name,
@ -507,10 +511,14 @@ class ProjectViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
tmp_dir = backup.get_backup_dirname() tmp_dir = backup.get_backup_dirname()
backup_file = os.path.join(tmp_dir, filename) backup_file = os.path.join(tmp_dir, filename)
if os.path.isfile(backup_file): if os.path.isfile(backup_file):
return backup.import_project(request, filename=backup_file) return backup.import_project(
request,
settings.CVAT_QUEUES.IMPORT_DATA.value,
filename=backup_file,
)
return Response(data='No such file were uploaded', return Response(data='No such file were uploaded',
status=status.HTTP_400_BAD_REQUEST) status=status.HTTP_400_BAD_REQUEST)
return backup.import_project(request) return backup.import_project(request, settings.CVAT_QUEUES.IMPORT_DATA.value)
return Response(data='Unknown upload was finished', return Response(data='Unknown upload was finished',
status=status.HTTP_400_BAD_REQUEST) status=status.HTTP_400_BAD_REQUEST)
@ -933,7 +941,8 @@ class TaskViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
return _import_annotations( return _import_annotations(
request=request, request=request,
filename=annotation_file, filename=annotation_file,
rq_id="{}@/api/tasks/{}/annotations/upload".format(request.user, self._object.pk), rq_id=(f"import:annotations-for-task.id{self._object.pk}-"
f"in-{format_name.replace(' ', '_')}-by-{request.user}"),
rq_func=dm.task.import_task_annotations, rq_func=dm.task.import_task_annotations,
pk=self._object.pk, pk=self._object.pk,
format_name=format_name, format_name=format_name,
@ -972,7 +981,7 @@ class TaskViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
# the value specified by the user or it's default value from the database # the value specified by the user or it's default value from the database
if 'stop_frame' not in serializer.validated_data: if 'stop_frame' not in serializer.validated_data:
data['stop_frame'] = None data['stop_frame'] = None
task.create(self._object.id, data) task.create(self._object.id, data, request.user)
return Response(serializer.data, status=status.HTTP_202_ACCEPTED) return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
elif self.action == 'import_backup': elif self.action == 'import_backup':
filename = request.query_params.get("filename", "") filename = request.query_params.get("filename", "")
@ -980,10 +989,14 @@ class TaskViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
tmp_dir = backup.get_backup_dirname() tmp_dir = backup.get_backup_dirname()
backup_file = os.path.join(tmp_dir, filename) backup_file = os.path.join(tmp_dir, filename)
if os.path.isfile(backup_file): if os.path.isfile(backup_file):
return backup.import_task(request, filename=backup_file) return backup.import_task(
request,
settings.CVAT_QUEUES.IMPORT_DATA.value,
filename=backup_file,
)
return Response(data='No such file were uploaded', return Response(data='No such file were uploaded',
status=status.HTTP_400_BAD_REQUEST) status=status.HTTP_400_BAD_REQUEST)
return backup.import_task(request) return backup.import_task(request, settings.CVAT_QUEUES.IMPORT_DATA.value)
return Response(data='Unknown upload was finished', return Response(data='Unknown upload was finished',
status=status.HTTP_400_BAD_REQUEST) status=status.HTTP_400_BAD_REQUEST)
@ -1155,16 +1168,17 @@ class TaskViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
return Response(data="Exporting annotations from a task without data is not allowed", return Response(data="Exporting annotations from a task without data is not allowed",
status=status.HTTP_400_BAD_REQUEST) status=status.HTTP_400_BAD_REQUEST)
elif request.method == 'POST' or request.method == 'OPTIONS': elif request.method == 'POST' or request.method == 'OPTIONS':
format_name = request.query_params.get('format', '')
return self.import_annotations( return self.import_annotations(
request=request, request=request,
pk=pk, pk=pk,
db_obj=self._object, db_obj=self._object,
import_func=_import_annotations, import_func=_import_annotations,
rq_func=dm.task.import_task_annotations, rq_func=dm.task.import_task_annotations,
rq_id = "{}@/api/tasks/{}/annotations/upload".format(request.user, pk) rq_id = f"import:annotations-for-task.id{pk}-in-{format_name.replace(' ', '_')}-by-{request.user}"
) )
elif request.method == 'PUT': elif request.method == 'PUT':
format_name = request.query_params.get('format') format_name = request.query_params.get('format', '')
if format_name: if format_name:
use_settings = strtobool(str(request.query_params.get('use_default_location', True))) use_settings = strtobool(str(request.query_params.get('use_default_location', True)))
conv_mask_to_poly = strtobool(request.query_params.get('conv_mask_to_poly', 'True')) conv_mask_to_poly = strtobool(request.query_params.get('conv_mask_to_poly', 'True'))
@ -1174,7 +1188,7 @@ class TaskViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
) )
return _import_annotations( return _import_annotations(
request=request, request=request,
rq_id="{}@/api/tasks/{}/annotations/upload".format(request.user, pk), rq_id = f"import:annotations-for-task.id{pk}-in-{format_name.replace(' ', '_')}-by-{request.user}",
rq_func=dm.task.import_task_annotations, rq_func=dm.task.import_task_annotations,
pk=pk, pk=pk,
format_name=format_name, format_name=format_name,
@ -1226,7 +1240,10 @@ class TaskViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
@action(detail=True, methods=['GET'], serializer_class=RqStatusSerializer) @action(detail=True, methods=['GET'], serializer_class=RqStatusSerializer)
def status(self, request, pk): def status(self, request, pk):
self.get_object() # force to call check_object_permissions self.get_object() # force to call check_object_permissions
response = self._get_rq_response(queue="default", job_id=f"/api/tasks/{pk}") response = self._get_rq_response(
queue=settings.CVAT_QUEUES.IMPORT_DATA.value,
job_id=f"create:task.id{pk}-by-{request.user}"
)
serializer = RqStatusSerializer(data=response) serializer = RqStatusSerializer(data=response)
if serializer.is_valid(raise_exception=True): if serializer.is_valid(raise_exception=True):
@ -1437,7 +1454,8 @@ class JobViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
return _import_annotations( return _import_annotations(
request=request, request=request,
filename=annotation_file, filename=annotation_file,
rq_id="{}@/api/jobs/{}/annotations/upload".format(request.user, self._object.pk), rq_id=(f"import:annotations-for-job.id{self._object.pk}-"
f"in-{format_name.replace(' ', '_')}-by-{request.user}"),
rq_func=dm.task.import_job_annotations, rq_func=dm.task.import_job_annotations,
pk=self._object.pk, pk=self._object.pk,
format_name=format_name, format_name=format_name,
@ -1544,13 +1562,15 @@ class JobViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
) )
elif request.method == 'POST' or request.method == 'OPTIONS': elif request.method == 'POST' or request.method == 'OPTIONS':
format_name = request.query_params.get('format', '')
return self.import_annotations( return self.import_annotations(
request=request, request=request,
pk=pk, pk=pk,
db_obj=self._object.segment.task, db_obj=self._object.segment.task,
import_func=_import_annotations, import_func=_import_annotations,
rq_func=dm.task.import_job_annotations, rq_func=dm.task.import_job_annotations,
rq_id = "{}@/api/jobs/{}/annotations/upload".format(request.user, pk) rq_id=(f"import:annotations-for-job.id{self._object.pk}-"
f"in-{format_name.replace(' ', '_')}-by-{request.user}"),
) )
elif request.method == 'PUT': elif request.method == 'PUT':
@ -1564,7 +1584,8 @@ class JobViewSet(viewsets.GenericViewSet, mixins.ListModelMixin,
) )
return _import_annotations( return _import_annotations(
request=request, request=request,
rq_id="{}@/api/jobs/{}/annotations/upload".format(request.user, pk), rq_id=(f"import:annotations-for-job.id{pk}-"
f"in-{format_name.replace(' ', '_')}-by-{request.user}"),
rq_func=dm.task.import_job_annotations, rq_func=dm.task.import_job_annotations,
pk=pk, pk=pk,
format_name=format_name, format_name=format_name,
@ -2248,7 +2269,7 @@ def _import_annotations(request, rq_id, rq_func, pk, format_name,
elif not format_desc.ENABLED: elif not format_desc.ENABLED:
return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)
queue = django_rq.get_queue("default") queue = django_rq.get_queue(settings.CVAT_QUEUES.IMPORT_DATA.value)
rq_job = queue.fetch_job(rq_id) rq_job = queue.fetch_job(rq_id)
if not rq_job: if not rq_job:
@ -2329,7 +2350,7 @@ def _export_annotations(db_instance, rq_id, request, format_name, action, callba
elif not format_desc.ENABLED: elif not format_desc.ENABLED:
return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)
queue = django_rq.get_queue("default") queue = django_rq.get_queue(settings.CVAT_QUEUES.EXPORT_DATA.value)
rq_job = queue.fetch_job(rq_id) rq_job = queue.fetch_job(rq_id)
if rq_job: if rq_job:
@ -2420,7 +2441,7 @@ def _import_project_dataset(request, rq_id, rq_func, pk, format_name, filename=N
elif not format_desc.ENABLED: elif not format_desc.ENABLED:
return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)
queue = django_rq.get_queue("default") queue = django_rq.get_queue(settings.CVAT_QUEUES.IMPORT_DATA.value)
rq_job = queue.fetch_job(rq_id) rq_job = queue.fetch_job(rq_id)
if not rq_job: if not rq_job:

@ -364,8 +364,7 @@ class LambdaFunction:
class LambdaQueue: class LambdaQueue:
def _get_queue(self): def _get_queue(self):
QUEUE_NAME = "low" return django_rq.get_queue(settings.CVAT_QUEUES.AUTO_ANNOTATION.value)
return django_rq.get_queue(QUEUE_NAME)
def get_jobs(self): def get_jobs(self):
queue = self._get_queue() queue = self._get_queue()

@ -10,6 +10,7 @@ import json
import django_rq import django_rq
import requests import requests
from django.dispatch import Signal, receiver from django.dispatch import Signal, receiver
from django.conf import settings
from cvat.apps.engine.models import Project from cvat.apps.engine.models import Project
from cvat.apps.engine.serializers import BasicUserSerializer from cvat.apps.engine.serializers import BasicUserSerializer
@ -75,7 +76,7 @@ def add_to_queue(webhook, payload, redelivery=False):
response="", response="",
) )
queue = django_rq.get_queue("webhooks") queue = django_rq.get_queue(settings.CVAT_QUEUES.WEBHOOKS.value)
queue.enqueue_call(func=send_webhook, args=(webhook, payload, delivery)) queue.enqueue_call(func=send_webhook, args=(webhook, payload, delivery))
return delivery return delivery

@ -15,6 +15,7 @@ For the full list of settings and their values, see
https://docs.djangoproject.com/en/2.0/ref/settings/ https://docs.djangoproject.com/en/2.0/ref/settings/
""" """
from enum import Enum
import os import os
import sys import sys
import fcntl import fcntl
@ -296,20 +297,32 @@ OLD_PASSWORD_FIELD_ENABLED = True
# Django-RQ # Django-RQ
# https://github.com/rq/django-rq # https://github.com/rq/django-rq
class CVAT_QUEUES(Enum):
IMPORT_DATA = 'import'
EXPORT_DATA = 'export'
AUTO_ANNOTATION = 'annotation'
WEBHOOKS = 'webhooks'
RQ_QUEUES = { RQ_QUEUES = {
'default': { CVAT_QUEUES.IMPORT_DATA.value: {
'HOST': 'localhost',
'PORT': 6379,
'DB': 0,
'DEFAULT_TIMEOUT': '4h'
},
CVAT_QUEUES.EXPORT_DATA.value: {
'HOST': 'localhost', 'HOST': 'localhost',
'PORT': 6379, 'PORT': 6379,
'DB': 0, 'DB': 0,
'DEFAULT_TIMEOUT': '4h' 'DEFAULT_TIMEOUT': '4h'
}, },
'low': { CVAT_QUEUES.AUTO_ANNOTATION.value: {
'HOST': 'localhost', 'HOST': 'localhost',
'PORT': 6379, 'PORT': 6379,
'DB': 0, 'DB': 0,
'DEFAULT_TIMEOUT': '24h' 'DEFAULT_TIMEOUT': '24h'
}, },
'webhooks': { CVAT_QUEUES.WEBHOOKS.value: {
'HOST': 'localhost', 'HOST': 'localhost',
'PORT': 6379, 'PORT': 6379,
'DB': 0, 'DB': 0,

@ -16,7 +16,6 @@ services:
CLAM_AV: CLAM_AV:
INSTALL_SOURCES: INSTALL_SOURCES:
CVAT_DEBUG_ENABLED: CVAT_DEBUG_ENABLED:
command: '-c supervisord/server.conf'
environment: environment:
CVAT_DEBUG_ENABLED: '${CVAT_DEBUG_ENABLED:-no}' CVAT_DEBUG_ENABLED: '${CVAT_DEBUG_ENABLED:-no}'
CVAT_DEBUG_PORT: '9090' CVAT_DEBUG_PORT: '9090'
@ -25,8 +24,7 @@ services:
ports: ports:
- '9090:9090' - '9090:9090'
cvat_worker_default: cvat_worker_export:
command: -c supervisord/worker.default.conf
environment: environment:
# For debugging, make sure to set 1 process # For debugging, make sure to set 1 process
# Due to the supervisord specifics, the extra processes will fail and # Due to the supervisord specifics, the extra processes will fail and
@ -37,8 +35,18 @@ services:
ports: ports:
- '9092:9092' - '9092:9092'
cvat_worker_low: cvat_worker_import:
command: -c supervisord/worker.low.conf environment:
# For debugging, make sure to set 1 process
# Due to the supervisord specifics, the extra processes will fail and
# after few attempts supervisord will give up restarting, leaving only 1 process
# NUMPROCS: 1
CVAT_DEBUG_ENABLED: '${CVAT_DEBUG_ENABLED:-no}'
CVAT_DEBUG_PORT: '9093'
ports:
- '9093:9093'
cvat_worker_annotation:
environment: environment:
# For debugging, make sure to set 1 process # For debugging, make sure to set 1 process
# Due to the supervisord specifics, the extra processes will fail and # Due to the supervisord specifics, the extra processes will fail and

@ -82,20 +82,39 @@ services:
networks: networks:
- cvat - cvat
cvat_worker_default: cvat_worker_import:
container_name: cvat_worker_default container_name: cvat_worker_import
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'
no_proxy: elasticsearch,kibana,logstash,nuclio,opa,${no_proxy:-}
NUMPROCS: 2
command: -c supervisord/worker.import.conf
volumes:
- cvat_data:/home/django/data
- cvat_keys:/home/django/keys
- cvat_logs:/home/django/logs
networks:
- cvat
cvat_worker_export:
container_name: cvat_worker_export
image: cvat/server:${CVAT_VERSION:-dev} image: cvat/server:${CVAT_VERSION:-dev}
restart: always restart: always
depends_on: depends_on:
- cvat_redis - cvat_redis
- cvat_db - cvat_db
- cvat_opa
environment: environment:
CVAT_REDIS_HOST: 'cvat_redis' CVAT_REDIS_HOST: 'cvat_redis'
CVAT_POSTGRES_HOST: 'cvat_db' CVAT_POSTGRES_HOST: 'cvat_db'
no_proxy: elasticsearch,kibana,logstash,nuclio,opa,${no_proxy:-} no_proxy: elasticsearch,kibana,logstash,nuclio,opa,${no_proxy:-}
NUMPROCS: 2 NUMPROCS: 2
command: -c supervisord/worker.default.conf command: -c supervisord/worker.export.conf
volumes: volumes:
- cvat_data:/home/django/data - cvat_data:/home/django/data
- cvat_keys:/home/django/keys - cvat_keys:/home/django/keys
@ -103,8 +122,8 @@ services:
networks: networks:
- cvat - cvat
cvat_worker_low: cvat_worker_annotation:
container_name: cvat_worker_low container_name: cvat_worker_annotation
image: cvat/server:${CVAT_VERSION:-dev} image: cvat/server:${CVAT_VERSION:-dev}
restart: always restart: always
depends_on: depends_on:
@ -116,7 +135,7 @@ services:
CVAT_POSTGRES_HOST: 'cvat_db' CVAT_POSTGRES_HOST: 'cvat_db'
no_proxy: elasticsearch,kibana,logstash,nuclio,opa,${no_proxy:-} no_proxy: elasticsearch,kibana,logstash,nuclio,opa,${no_proxy:-}
NUMPROCS: 1 NUMPROCS: 1
command: -c supervisord/worker.low.conf command: -c supervisord/worker.annotation.conf
volumes: volumes:
- cvat_data:/home/django/data - cvat_data:/home/django/data
- cvat_keys:/home/django/keys - cvat_keys:/home/django/keys

@ -15,7 +15,7 @@ type: application
# This is the chart version. This version number should be incremented each time you make changes # This is the chart version. This version number should be incremented each time you make changes
# to the chart and its templates, including the app version. # to the chart and its templates, including the app version.
# Versions are expected to follow Semantic Versioning (https://semver.org/) # Versions are expected to follow Semantic Versioning (https://semver.org/)
version: 0.5.2 version: 0.6.0
# This is the version number of the application being deployed. This version number should be # This is the version number of the application being deployed. This version number should be
# incremented each time you make changes to the application. Versions are not expected to # incremented each time you make changes to the application. Versions are not expected to

@ -0,0 +1,170 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: {{ .Release.Name }}-backend-worker-annotation
namespace: {{ .Release.Namespace }}
labels:
app: cvat-app
tier: backend
component: worker-annotation
{{- include "cvat.labels" . | nindent 4 }}
{{- with .Values.cvat.backend.worker.annotation.labels }}
{{- toYaml . | nindent 4 }}
{{- end }}
{{- with .Values.cvat.backend.worker.annotation.annotations }}
annotations:
{{- toYaml . | nindent 4 }}
{{- end }}
spec:
replicas: {{ .Values.cvat.backend.worker.annotation.replicas }}
strategy:
type: Recreate
selector:
matchLabels:
{{- include "cvat.labels" . | nindent 6 }}
{{- with .Values.cvat.backend.worker.annotation.labels }}
{{- toYaml . | nindent 6 }}
{{- end }}
app: cvat-app
tier: backend
component: worker-annotation
template:
metadata:
labels:
app: cvat-app
tier: backend
component: worker-annotation
{{- include "cvat.labels" . | nindent 8 }}
{{- with .Values.cvat.backend.worker.annotation.labels }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.cvat.backend.worker.annotation.annotations }}
annotations:
{{- toYaml . | nindent 8 }}
{{- end }}
spec:
containers:
- name: cvat-app-backend-worker-annotation-container
image: {{ .Values.cvat.backend.image }}:{{ .Values.cvat.backend.tag }}
imagePullPolicy: {{ .Values.cvat.backend.imagePullPolicy }}
{{- with .Values.cvat.backend.worker.annotation.resources }}
resources:
{{- toYaml . | nindent 12 }}
{{- end }}
args: ["-c", "supervisord/worker.annotation.conf"]
env:
{{- if .Values.redis.enabled }}
- name: CVAT_REDIS_HOST
value: "{{ .Release.Name }}-redis-master"
{{- else }}
- name: CVAT_REDIS_HOST
value: "{{ .Values.redis.external.host }}"
{{- end }}
- name: CVAT_REDIS_PASSWORD
valueFrom:
secretKeyRef:
name: "{{ tpl (.Values.redis.secret.name) . }}"
key: redis-password
{{- if .Values.postgresql.enabled }}
- name: CVAT_POSTGRES_HOST
value: "{{ .Release.Name }}-postgresql"
- name: CVAT_POSTGRES_PORT
value: "{{ .Values.postgresql.service.ports.postgresql }}"
{{- else }}
- name: CVAT_POSTGRES_HOST
value: "{{ .Values.postgresql.external.host }}"
- name: CVAT_POSTGRES_PORT
value: "{{ .Values.postgresql.external.port }}"
{{- end }}
- name: CVAT_POSTGRES_USER
valueFrom:
secretKeyRef:
name: "{{ tpl (.Values.postgresql.secret.name) . }}"
key: username
- name: CVAT_POSTGRES_DBNAME
valueFrom:
secretKeyRef:
name: "{{ tpl (.Values.postgresql.secret.name) . }}"
key: database
- name: CVAT_POSTGRES_PASSWORD
valueFrom:
secretKeyRef:
name: "{{ tpl (.Values.postgresql.secret.name) . }}"
key: password
{{- if .Values.nuclio }}
- name: CVAT_SERVERLESS
value: "1"
- name: CVAT_NUCLIO_HOST
value: "{{ .Release.Name }}-nuclio-dashboard"
{{- end }}
{{- with .Values.cvat.backend.worker.annotation.additionalEnv }}
{{- toYaml . | nindent 10 }}
{{- end }}
volumeMounts:
- mountPath: /home/django/data
name: cvat-backend-data
subPath: data
- mountPath: /home/django/keys
name: cvat-backend-data
subPath: keys
- mountPath: /home/django/logs
name: cvat-backend-data
subPath: logs
- mountPath: /home/django/models
name: cvat-backend-data
subPath: models
- mountPath: /home/django/tmp_storage
name: cvat-backend-data
subPath: tmp_storage
{{- with .Values.cvat.backend.worker.annotation.additionalVolumeMounts }}
{{- toYaml . | nindent 10 }}
{{- end }}
initContainers:
{{- if .Values.cvat.backend.permissionFix.enabled }}
- name: user-data-permission-fix
image: busybox
command: ["/bin/chmod", "-R", "777", "/home/django"]
{{- with .Values.cvat.backend.worker.annotation.resources }}
resources:
{{- toYaml . | nindent 12 }}
{{- end }}
volumeMounts:
{{- if .Values.cvat.backend.defaultStorage.enabled }}
- mountPath: /home/django/data
name: cvat-backend-data
subPath: data
- mountPath: /home/django/keys
name: cvat-backend-data
subPath: keys
- mountPath: /home/django/logs
name: cvat-backend-data
subPath: logs
- mountPath: /home/django/models
name: cvat-backend-data
subPath: models
{{- end }}
{{- with .Values.cvat.backend.worker.annotation.additionalVolumeMounts }}
{{- toYaml . | nindent 10 }}
{{- end }}
{{- end }}
{{- with .Values.cvat.backend.worker.annotation.affinity }}
affinity:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.cvat.backend.worker.annotation.tolerations }}
tolerations:
{{- toYaml . | nindent 8 }}
{{- end }}
volumes:
{{- if .Values.cvat.backend.defaultStorage.enabled }}
- name: cvat-backend-data
persistentVolumeClaim:
claimName: "{{ .Release.Name }}-backend-data"
{{- end }}
{{- with .Values.cvat.backend.worker.annotation.additionalVolumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.imagePullSecrets }}
imagePullSecrets:
{{- toYaml . | nindent 8 }}
{{- end }}

@ -1,57 +1,57 @@
apiVersion: apps/v1 apiVersion: apps/v1
kind: Deployment kind: Deployment
metadata: metadata:
name: {{ .Release.Name }}-backend-worker-low name: {{ .Release.Name }}-backend-worker-export
namespace: {{ .Release.Namespace }} namespace: {{ .Release.Namespace }}
labels: labels:
app: cvat-app app: cvat-app
tier: backend tier: backend
component: worker-low component: worker-export
{{- include "cvat.labels" . | nindent 4 }} {{- include "cvat.labels" . | nindent 4 }}
{{- with .Values.cvat.backend.worker.low.labels }} {{- with .Values.cvat.backend.worker.export.labels }}
{{- toYaml . | nindent 4 }} {{- toYaml . | nindent 4 }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.low.annotations }} {{- with .Values.cvat.backend.worker.export.annotations }}
annotations: annotations:
{{- toYaml . | nindent 4 }} {{- toYaml . | nindent 4 }}
{{- end }} {{- end }}
spec: spec:
replicas: {{ .Values.cvat.backend.worker.low.replicas }} replicas: {{ .Values.cvat.backend.worker.export.replicas }}
strategy: strategy:
type: Recreate type: Recreate
selector: selector:
matchLabels: matchLabels:
{{- include "cvat.labels" . | nindent 6 }} {{- include "cvat.labels" . | nindent 6 }}
{{- with .Values.cvat.backend.worker.low.labels }} {{- with .Values.cvat.backend.worker.export.labels }}
{{- toYaml . | nindent 6 }} {{- toYaml . | nindent 6 }}
{{- end }} {{- end }}
app: cvat-app app: cvat-app
tier: backend tier: backend
component: worker-low component: worker-export
template: template:
metadata: metadata:
labels: labels:
app: cvat-app app: cvat-app
tier: backend tier: backend
component: worker-low component: worker-export
{{- include "cvat.labels" . | nindent 8 }} {{- include "cvat.labels" . | nindent 8 }}
{{- with .Values.cvat.backend.worker.low.labels }} {{- with .Values.cvat.backend.worker.export.labels }}
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.low.annotations }} {{- with .Values.cvat.backend.worker.export.annotations }}
annotations: annotations:
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
spec: spec:
containers: containers:
- name: cvat-app-backend-worker-low-container - name: cvat-app-backend-worker-export-container
image: {{ .Values.cvat.backend.image }}:{{ .Values.cvat.backend.tag }} image: {{ .Values.cvat.backend.image }}:{{ .Values.cvat.backend.tag }}
imagePullPolicy: {{ .Values.cvat.backend.imagePullPolicy }} imagePullPolicy: {{ .Values.cvat.backend.imagePullPolicy }}
{{- with .Values.cvat.backend.worker.low.resources }} {{- with .Values.cvat.backend.worker.export.resources }}
resources: resources:
{{- toYaml . | nindent 12 }} {{- toYaml . | nindent 12 }}
{{- end }} {{- end }}
args: ["-c", "supervisord/worker.low.conf"] args: ["-c", "supervisord/worker.export.conf"]
env: env:
{{- if .Values.redis.enabled }} {{- if .Values.redis.enabled }}
- name: CVAT_REDIS_HOST - name: CVAT_REDIS_HOST
@ -65,7 +65,7 @@ spec:
secretKeyRef: secretKeyRef:
name: "{{ tpl (.Values.redis.secret.name) . }}" name: "{{ tpl (.Values.redis.secret.name) . }}"
key: redis-password key: redis-password
{{- if .Values.postgresql.enabled }} {{- if .Values.postgresql.enabled }}
- name: CVAT_POSTGRES_HOST - name: CVAT_POSTGRES_HOST
value: "{{ .Release.Name }}-postgresql" value: "{{ .Release.Name }}-postgresql"
- name: CVAT_POSTGRES_PORT - name: CVAT_POSTGRES_PORT
@ -97,11 +97,9 @@ spec:
- name: CVAT_NUCLIO_HOST - name: CVAT_NUCLIO_HOST
value: "{{ .Release.Name }}-nuclio-dashboard" value: "{{ .Release.Name }}-nuclio-dashboard"
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.low.additionalEnv }} {{- with .Values.cvat.backend.worker.export.additionalEnv }}
{{- toYaml . | nindent 10 }} {{- toYaml . | nindent 10 }}
{{- end }} {{- end }}
ports:
- containerPort: 8080
volumeMounts: volumeMounts:
- mountPath: /home/django/data - mountPath: /home/django/data
name: cvat-backend-data name: cvat-backend-data
@ -118,7 +116,7 @@ spec:
- mountPath: /home/django/tmp_storage - mountPath: /home/django/tmp_storage
name: cvat-backend-data name: cvat-backend-data
subPath: tmp_storage subPath: tmp_storage
{{- with .Values.cvat.backend.worker.low.additionalVolumeMounts }} {{- with .Values.cvat.backend.worker.export.additionalVolumeMounts }}
{{- toYaml . | nindent 10 }} {{- toYaml . | nindent 10 }}
{{- end }} {{- end }}
initContainers: initContainers:
@ -126,7 +124,7 @@ spec:
- name: user-data-permission-fix - name: user-data-permission-fix
image: busybox image: busybox
command: ["/bin/chmod", "-R", "777", "/home/django"] command: ["/bin/chmod", "-R", "777", "/home/django"]
{{- with .Values.cvat.backend.worker.low.resources }} {{- with .Values.cvat.backend.worker.export.resources }}
resources: resources:
{{- toYaml . | nindent 12 }} {{- toYaml . | nindent 12 }}
{{- end }} {{- end }}
@ -145,15 +143,15 @@ spec:
name: cvat-backend-data name: cvat-backend-data
subPath: models subPath: models
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.low.additionalVolumeMounts }} {{- with .Values.cvat.backend.worker.export.additionalVolumeMounts }}
{{- toYaml . | nindent 10 }} {{- toYaml . | nindent 10 }}
{{- end }} {{- end }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.low.affinity }} {{- with .Values.cvat.backend.worker.export.affinity }}
affinity: affinity:
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.low.tolerations }} {{- with .Values.cvat.backend.worker.export.tolerations }}
tolerations: tolerations:
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
@ -163,7 +161,7 @@ spec:
persistentVolumeClaim: persistentVolumeClaim:
claimName: "{{ .Release.Name }}-backend-data" claimName: "{{ .Release.Name }}-backend-data"
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.low.additionalVolumes }} {{- with .Values.cvat.backend.worker.export.additionalVolumes }}
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
{{- with .Values.imagePullSecrets }} {{- with .Values.imagePullSecrets }}

@ -1,57 +1,57 @@
apiVersion: apps/v1 apiVersion: apps/v1
kind: Deployment kind: Deployment
metadata: metadata:
name: {{ .Release.Name }}-backend-worker-default name: {{ .Release.Name }}-backend-worker-import
namespace: {{ .Release.Namespace }} namespace: {{ .Release.Namespace }}
labels: labels:
app: cvat-app app: cvat-app
tier: backend tier: backend
component: worker-default component: worker-import
{{- include "cvat.labels" . | nindent 4 }} {{- include "cvat.labels" . | nindent 4 }}
{{- with .Values.cvat.backend.worker.default.labels }} {{- with .Values.cvat.backend.worker.import.labels }}
{{- toYaml . | nindent 4 }} {{- toYaml . | nindent 4 }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.default.annotations }} {{- with .Values.cvat.backend.worker.import.annotations }}
annotations: annotations:
{{- toYaml . | nindent 4 }} {{- toYaml . | nindent 4 }}
{{- end }} {{- end }}
spec: spec:
replicas: {{ .Values.cvat.backend.worker.default.replicas }} replicas: {{ .Values.cvat.backend.worker.import.replicas }}
strategy: strategy:
type: Recreate type: Recreate
selector: selector:
matchLabels: matchLabels:
{{- include "cvat.labels" . | nindent 6 }} {{- include "cvat.labels" . | nindent 6 }}
{{- with .Values.cvat.backend.worker.default.labels }} {{- with .Values.cvat.backend.worker.import.labels }}
{{- toYaml . | nindent 6 }} {{- toYaml . | nindent 6 }}
{{- end }} {{- end }}
app: cvat-app-worker-default app: cvat-app
tier: backend tier: backend
component: worker-default component: worker-import
template: template:
metadata: metadata:
labels: labels:
app: cvat-app-worker-default app: cvat-app
tier: backend tier: backend
component: worker-default component: worker-import
{{- include "cvat.labels" . | nindent 8 }} {{- include "cvat.labels" . | nindent 8 }}
{{- with .Values.cvat.backend.worker.default.labels }} {{- with .Values.cvat.backend.worker.import.labels }}
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.default.annotations }} {{- with .Values.cvat.backend.worker.import.annotations }}
annotations: annotations:
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
spec: spec:
containers: containers:
- name: cvat-app-backend-worker-default-container - name: cvat-app-backend-worker-import-container
image: {{ .Values.cvat.backend.image }}:{{ .Values.cvat.backend.tag }} image: {{ .Values.cvat.backend.image }}:{{ .Values.cvat.backend.tag }}
imagePullPolicy: {{ .Values.cvat.backend.imagePullPolicy }} imagePullPolicy: {{ .Values.cvat.backend.imagePullPolicy }}
{{- with .Values.cvat.backend.worker.default.resources }} {{- with .Values.cvat.backend.worker.import.resources }}
resources: resources:
{{- toYaml . | nindent 12 }} {{- toYaml . | nindent 12 }}
{{- end }} {{- end }}
args: ["-c", "supervisord/worker.default.conf"] args: ["-c", "supervisord/worker.import.conf"]
env: env:
{{- if .Values.redis.enabled }} {{- if .Values.redis.enabled }}
- name: CVAT_REDIS_HOST - name: CVAT_REDIS_HOST
@ -97,11 +97,9 @@ spec:
- name: CVAT_NUCLIO_HOST - name: CVAT_NUCLIO_HOST
value: "{{ .Release.Name }}-nuclio-dashboard" value: "{{ .Release.Name }}-nuclio-dashboard"
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.default.additionalEnv }} {{- with .Values.cvat.backend.worker.import.additionalEnv }}
{{- toYaml . | nindent 10 }} {{- toYaml . | nindent 10 }}
{{- end }} {{- end }}
ports:
- containerPort: 8080
volumeMounts: volumeMounts:
- mountPath: /home/django/data - mountPath: /home/django/data
name: cvat-backend-data name: cvat-backend-data
@ -118,7 +116,7 @@ spec:
- mountPath: /home/django/tmp_storage - mountPath: /home/django/tmp_storage
name: cvat-backend-data name: cvat-backend-data
subPath: tmp_storage subPath: tmp_storage
{{- with .Values.cvat.backend.worker.default.additionalVolumeMounts }} {{- with .Values.cvat.backend.worker.import.additionalVolumeMounts }}
{{- toYaml . | nindent 10 }} {{- toYaml . | nindent 10 }}
{{- end }} {{- end }}
initContainers: initContainers:
@ -126,7 +124,7 @@ spec:
- name: user-data-permission-fix - name: user-data-permission-fix
image: busybox image: busybox
command: ["/bin/chmod", "-R", "777", "/home/django"] command: ["/bin/chmod", "-R", "777", "/home/django"]
{{- with .Values.cvat.backend.worker.default.resources }} {{- with .Values.cvat.backend.worker.import.resources }}
resources: resources:
{{- toYaml . | nindent 12 }} {{- toYaml . | nindent 12 }}
{{- end }} {{- end }}
@ -145,15 +143,15 @@ spec:
name: cvat-backend-data name: cvat-backend-data
subPath: models subPath: models
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.default.additionalVolumeMounts }} {{- with .Values.cvat.backend.worker.import.additionalVolumeMounts }}
{{- toYaml . | nindent 10 }} {{- toYaml . | nindent 10 }}
{{- end }} {{- end }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.default.affinity }} {{- with .Values.cvat.backend.worker.import.affinity }}
affinity: affinity:
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.default.tolerations }} {{- with .Values.cvat.backend.worker.import.tolerations }}
tolerations: tolerations:
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
@ -163,7 +161,7 @@ spec:
persistentVolumeClaim: persistentVolumeClaim:
claimName: "{{ .Release.Name }}-backend-data" claimName: "{{ .Release.Name }}-backend-data"
{{- end }} {{- end }}
{{- with .Values.cvat.backend.worker.default.additionalVolumes }} {{- with .Values.cvat.backend.worker.import.additionalVolumes }}
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
{{- with .Values.imagePullSecrets }} {{- with .Values.imagePullSecrets }}

@ -32,7 +32,7 @@ cvat:
additionalVolumes: [] additionalVolumes: []
additionalVolumeMounts: [] additionalVolumeMounts: []
worker: worker:
default: export:
replicas: 2 replicas: 2
labels: {} labels: {}
annotations: {} annotations: {}
@ -42,7 +42,17 @@ cvat:
additionalEnv: [] additionalEnv: []
additionalVolumes: [] additionalVolumes: []
additionalVolumeMounts: [] additionalVolumeMounts: []
low: import:
replicas: 2
labels: {}
annotations: {}
resources: {}
affinity: {}
tolerations: []
additionalEnv: []
additionalVolumes: []
additionalVolumeMounts: []
annotation:
replicas: 1 replicas: 1
labels: {} labels: {}
annotations: {} annotations: {}

@ -22,16 +22,22 @@ command=bash -c "rm /tmp/ssh-agent.sock -f && /usr/bin/ssh-agent -d -a /tmp/ssh-
priority=1 priority=1
autorestart=true autorestart=true
[program:rqworker_default] [program:rqworker_export]
command=%(ENV_HOME)s/wait-for-it.sh %(ENV_CVAT_REDIS_HOST)s:6379 -t 0 -- bash -ic \ 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 default" "exec python3 %(ENV_HOME)s/manage.py rqworker -v 3 export"
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock" environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
numprocs=2 numprocs=2
process_name=rqworker_default_%(process_num)s process_name=rqworker_default_%(process_num)s
[program:rqworker_low]
command=%(ENV_HOME)s/wait-for-it.sh %(ENV_CVAT_REDIS_HOST)s:6379 -t 0 -- bash -ic \ 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 low" "exec python3 %(ENV_HOME)s/manage.py rqworker -v 3 import"
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
numprocs=2
process_name=rqworker_default_%(process_num)s
[program:rqworker_annotation]
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 annotation"
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock" environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
numprocs=1 numprocs=1
@ -49,7 +55,7 @@ numprocs=1
[program:rqscheduler] [program:rqscheduler]
command=%(ENV_HOME)s/wait-for-it.sh %(ENV_CVAT_REDIS_HOST)s:6379 -t 0 -- bash -ic \ command=%(ENV_HOME)s/wait-for-it.sh %(ENV_CVAT_REDIS_HOST)s:6379 -t 0 -- bash -ic \
"python3 /opt/venv/bin/rqscheduler --host %(ENV_CVAT_REDIS_HOST)s -i 30" "python3 /opt/venv/bin/rqscheduler --host %(ENV_CVAT_REDIS_HOST)s --password '%(ENV_CVAT_REDIS_PASSWORD)s' -i 30"
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock" environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
numprocs=1 numprocs=1

@ -22,18 +22,10 @@ command=bash -c "rm /tmp/ssh-agent.sock -f && /usr/bin/ssh-agent -d -a /tmp/ssh-
priority=1 priority=1
autorestart=true autorestart=true
[program:rqworker_default] [program:rqworker_annotation]
command=%(ENV_HOME)s/wait-for-it.sh %(ENV_CVAT_REDIS_HOST)s:6379 -t 0 -- bash -ic " \ 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 default \ exec python3 %(ENV_HOME)s/manage.py rqworker -v 3 annotation \
--worker-class cvat.rqworker.DefaultWorker \ --worker-class cvat.rqworker.DefaultWorker \
" "
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock" environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
numprocs=%(ENV_NUMPROCS)s numprocs=%(ENV_NUMPROCS)s
process_name=rqworker_default_%(process_num)s
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
[program:clamav_update]
command=bash -c "if [ \"${CLAM_AV}\" = 'yes' ]; then /usr/bin/freshclam -d \
-l %(ENV_HOME)s/logs/freshclam.log --foreground=true; fi"
numprocs=1

@ -0,0 +1,32 @@
[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_export]
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 export \
--worker-class cvat.rqworker.DefaultWorker \
"
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
numprocs=%(ENV_NUMPROCS)s
process_name=rqworker_export_%(process_num)s

@ -22,17 +22,18 @@ command=bash -c "rm /tmp/ssh-agent.sock -f && /usr/bin/ssh-agent -d -a /tmp/ssh-
priority=1 priority=1
autorestart=true autorestart=true
[program:rqworker_low] [program:rqworker_import]
command=%(ENV_HOME)s/wait-for-it.sh %(ENV_CVAT_REDIS_HOST)s:6379 -t 0 -- bash -ic " \ 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 low \ exec python3 %(ENV_HOME)s/manage.py rqworker -v 3 import \
--worker-class cvat.rqworker.DefaultWorker \ --worker-class cvat.rqworker.DefaultWorker \
" "
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock" environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
numprocs=%(ENV_NUMPROCS)s numprocs=%(ENV_NUMPROCS)s
process_name=rqworker_import_%(process_num)s
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"
[program:clamav_update] [program:clamav_update]
command=bash -c "if [ \"${CLAM_AV}\" = 'yes' ]; then /usr/bin/freshclam -d \ command=bash -c "if [ \"${CLAM_AV}\" = 'yes' ]; then /usr/bin/freshclam -d \
-l %(ENV_HOME)s/logs/freshclam.log --foreground=true; fi" -l %(ENV_HOME)s/logs/freshclam.log --foreground=true; fi"
numprocs=1 numprocs=1
environment=SSH_AUTH_SOCK="/tmp/ssh-agent.sock"

@ -1,5 +1,5 @@
services: services:
cvat_worker_default: cvat_worker_import:
volumes: volumes:
- ./tests/mounted_file_share:/home/django/share:rw - ./tests/mounted_file_share:/home/django/share:rw
cvat_server: cvat_server:

@ -6,7 +6,7 @@ cvat:
name: cvat-backend-data name: cvat-backend-data
subPath: share subPath: share
worker: worker:
default: import:
additionalVolumeMounts: additionalVolumeMounts:
- mountPath: /home/django/share - mountPath: /home/django/share
name: cvat-backend-data name: cvat-backend-data

Loading…
Cancel
Save