From 61bc772054359de7f5eebebd635b9a76d0f8ee92 Mon Sep 17 00:00:00 2001 From: mariomitte Date: Sat, 5 Sep 2026 08:15:30 +0200 Subject: [PATCH] fix: reduce memory usage for monthly ZIP archive flow Replace generated archive download response with FileResponse streaming so ZIP files are not fully loaded into process memory. Move monthly archive task output to a temporary file and upload that file to storage, and write invoice attachments into ZIP archives in chunks. Add Celery memory guard settings, a memory-aware base task hook, and periodic /tmp cleanup for prefixed archive files older than one day. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- backend/core/celery.py | 42 ++++++++++ backend/core/settings/base.py | 20 +++++ backend/modules/fleet/tasks.py | 99 ++++++++++++++++++---- backend/modules/fleet/views.py | 119 +++++++++++++++++++-------- backend/scripts/cleanup_tmp_files.sh | 24 ++++++ docker-compose.prod.yml | 8 +- 6 files changed, 260 insertions(+), 52 deletions(-) create mode 100644 backend/scripts/cleanup_tmp_files.sh diff --git a/backend/core/celery.py b/backend/core/celery.py index 4af78a7..5f09a95 100644 --- a/backend/core/celery.py +++ b/backend/core/celery.py @@ -1,7 +1,12 @@ # backend/core/celery.py import os +import gc +import logging from celery import Celery +from celery import Task + +logger = logging.getLogger(__name__) # Postavi Django settings modul os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings') @@ -11,5 +16,42 @@ app = Celery('core') # Koristi konfiguraciju iz settings.py s prefiksom 'CELERY_' app.config_from_object('django.conf:settings', namespace='CELERY') + +def _rss_memory_mb(): + try: + import resource + + usage_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + # Na Linuxu je ru_maxrss u KiB. + return round(float(usage_kb) / 1024.0, 2) + except (ImportError, AttributeError, OSError, ValueError): + return None + + +class MemoryAwareTask(Task): + abstract = True + + def __call__(self, *args, **kwargs): + gc.collect() + return super().__call__(*args, **kwargs) + + def after_return(self, status, retval, task_id, args, kwargs, einfo): + memory_before_gc = _rss_memory_mb() + gc.collect() + memory_after_gc = _rss_memory_mb() + if memory_before_gc is not None and memory_after_gc is not None: + logger.info( + "Task %s (%s) status=%s RSS prije/poslije GC: %.2fMB -> %.2fMB", + self.name, + task_id, + status, + memory_before_gc, + memory_after_gc, + ) + return super().after_return(status, retval, task_id, args, kwargs, einfo) + + +app.Task = MemoryAwareTask + # Automatski pronalazi taskove u svim 'modules.*' app.autodiscover_tasks() \ No newline at end of file diff --git a/backend/core/settings/base.py b/backend/core/settings/base.py index 8a72846..fb86453 100644 --- a/backend/core/settings/base.py +++ b/backend/core/settings/base.py @@ -154,6 +154,11 @@ REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0') CELERY_BROKER_URL = REDIS_URL CELERY_RESULT_BACKEND = REDIS_URL +CELERY_WORKER_PREFETCH_MULTIPLIER = int(os.environ.get('CELERY_WORKER_PREFETCH_MULTIPLIER', '1')) +CELERY_WORKER_MAX_TASKS_PER_CHILD = int(os.environ.get('CELERY_WORKER_MAX_TASKS_PER_CHILD', '20')) +CELERY_WORKER_MAX_MEMORY_PER_CHILD = int(os.environ.get('CELERY_WORKER_MAX_MEMORY_PER_CHILD', '350000')) +CELERY_TASK_ACKS_LATE = True +CELERY_TASK_REJECT_ON_WORKER_LOST = True # Ako želiš koristiti Redis kao brzi cache sustav unutar Djanga (izvrsno za ERP performanse) # Za ovo ti je potreban paket 'django-redis' u requirements.txt @@ -171,10 +176,18 @@ CACHES = { CELERY_ACCEPT_CONTENT = ['json'] CELERY_TASK_SERIALIZER = 'json' CELERY_BEAT_SCHEDULE = { + 'cleanup-expired-generated-archives-hourly': { + 'task': 'modules.fleet.tasks.cleanup_expired_generated_archives_task', + 'schedule': crontab(minute=5), + }, 'cleanup-expired-generated-pdfs-hourly': { 'task': 'modules.fleet.tasks.cleanup_expired_generated_pdfs_task', 'schedule': crontab(minute=0), }, + 'cleanup-app-tmp-daily': { + 'task': 'modules.fleet.tasks.cleanup_tmp_files_task', + 'schedule': crontab(hour=3, minute=15), + }, 'notify-upcoming-tasks-daily': { 'task': 'modules.task_management.tasks.notify_upcoming_tasks', 'schedule': crontab(hour=8, minute=0), @@ -190,6 +203,13 @@ STATIC_URL = '/static/' # (Opcionalno, ali preporučeno za ERP) Ako koristiš i medije (dokumente) MEDIA_URL = '/media/' MEDIA_ROOT = BASE_DIR / 'media' +APP_TMP_CLEANUP_DIR = os.environ.get('APP_TMP_CLEANUP_DIR', '/tmp') +APP_TMP_CLEANUP_MAX_AGE_HOURS = int(os.environ.get('APP_TMP_CLEANUP_MAX_AGE_HOURS', '24')) +APP_TMP_CLEANUP_PREFIXES = [ + prefix.strip() + for prefix in os.environ.get('APP_TMP_CLEANUP_PREFIXES', 'erp-fleet-archive-').split(',') + if prefix.strip() +] # ============================================================================== diff --git a/backend/modules/fleet/tasks.py b/backend/modules/fleet/tasks.py index 5ee21d8..4d2cf4b 100644 --- a/backend/modules/fleet/tasks.py +++ b/backend/modules/fleet/tasks.py @@ -5,6 +5,7 @@ from django.core.mail import EmailMessage from django.conf import settings from django.utils import timezone from django.core.files.base import ContentFile +from django.core.files import File import logging from io import BytesIO from io import StringIO @@ -13,6 +14,9 @@ import base64 import re import csv import mimetypes +import os +import tempfile +import zipfile from decimal import Decimal, InvalidOperation from smtplib import SMTPSenderRefused from datetime import timedelta @@ -838,10 +842,9 @@ def process_work_order_invoice_ocr(invoice_id): @shared_task def build_monthly_archive_cached_task(generated_archive_id): from .models import GeneratedFleetArchive - from .services import NotificationService from .views import ( - _build_monthly_service_tasks_archive_content, - _build_monthly_work_orders_archive_content, + _write_monthly_service_tasks_archive_entries, + _write_monthly_work_orders_archive_entries, _notify_monthly_archive_request, ) @@ -854,25 +857,39 @@ def build_monthly_archive_cached_task(generated_archive_id): if generated is None: return {"status": "failed", "error": "Generated archive record not found"} + tmp_dir = str(getattr(settings, 'APP_TMP_CLEANUP_DIR', '/tmp')) + with tempfile.NamedTemporaryFile( + mode='w+b', + delete=False, + dir=tmp_dir if os.path.isdir(tmp_dir) else None, + prefix='erp-fleet-archive-', + suffix='.zip', + ) as temp_zip: + temp_zip_path = temp_zip.name try: - if generated.archive_type == 'service_tasks': - archive_content = _build_monthly_service_tasks_archive_content( - user=generated.requested_by, - year=generated.year, - month=generated.month, - ) - else: - archive_content = _build_monthly_work_orders_archive_content( - user=generated.requested_by, - year=generated.year, - month=generated.month, - ) - - if not archive_content: + with zipfile.ZipFile(temp_zip_path, mode='w', compression=zipfile.ZIP_DEFLATED) as archive: + if generated.archive_type == 'service_tasks': + entries_written = _write_monthly_service_tasks_archive_entries( + archive, + user=generated.requested_by, + year=generated.year, + month=generated.month, + ) + else: + entries_written = _write_monthly_work_orders_archive_entries( + archive, + user=generated.requested_by, + year=generated.year, + month=generated.month, + ) + if entries_written == 0: raise ValueError('ZIP arhiva je prazna.') + if os.path.getsize(temp_zip_path) <= 0: + raise ValueError('ZIP arhiva je prazna.') filename = generated.filename or f"{generated.archive_type}-{generated.year}-{generated.month}.zip" - generated.file.save(filename, ContentFile(archive_content), save=False) + with open(temp_zip_path, 'rb') as temp_file: + generated.file.save(filename, File(temp_file), save=False) generated.status = 'ready' generated.error_message = '' generated.save(update_fields=['file', 'status', 'error_message', 'updated_at']) @@ -902,6 +919,12 @@ def build_monthly_archive_cached_task(generated_archive_id): ) logger.exception("Greška kod build_monthly_archive_cached_task: %s", exc) return {"status": "failed", "error": str(exc)} + finally: + try: + if os.path.exists(temp_zip_path): + os.remove(temp_zip_path) + except OSError: + logger.warning("Ne mogu obrisati privremenu ZIP datoteku: %s", temp_zip_path) @shared_task @@ -944,6 +967,46 @@ def cleanup_expired_generated_pdfs_task(): return {"deleted": deleted} +@shared_task +def cleanup_tmp_files_task(): + tmp_dir = str(getattr(settings, 'APP_TMP_CLEANUP_DIR', '/tmp')) + max_age_hours = int(getattr(settings, 'APP_TMP_CLEANUP_MAX_AGE_HOURS', 24)) + prefixes = tuple(getattr(settings, 'APP_TMP_CLEANUP_PREFIXES', ['erp-fleet-archive-'])) + + if max_age_hours <= 0: + raise ValueError('APP_TMP_CLEANUP_MAX_AGE_HOURS mora biti > 0.') + if not prefixes: + raise ValueError('APP_TMP_CLEANUP_PREFIXES ne smije biti prazan.') + if not os.path.isdir(tmp_dir): + return {'deleted': 0, 'tmp_dir': tmp_dir, 'reason': 'tmp-dir-not-found'} + + now = timezone.now().timestamp() + cutoff = now - (max_age_hours * 3600) + deleted = 0 + + for name in os.listdir(tmp_dir): + if not any(name.startswith(prefix) for prefix in prefixes): + continue + full_path = os.path.join(tmp_dir, name) + try: + stat_info = os.stat(full_path) + except FileNotFoundError: + continue + if stat_info.st_mtime > cutoff: + continue + if os.path.isdir(full_path): + try: + os.rmdir(full_path) + except OSError: + logger.warning("Preskačem %s: direktorij nije prazan ili je nedostupan.", full_path) + continue + else: + os.remove(full_path) + deleted += 1 + + return {'deleted': deleted, 'tmp_dir': tmp_dir, 'max_age_hours': max_age_hours} + + @shared_task def build_work_order_pdf_cached_task(generated_pdf_id): from .views import _build_work_order_pdf, _build_work_order_service_records_pdf diff --git a/backend/modules/fleet/views.py b/backend/modules/fleet/views.py index 24e77f7..b13d883 100644 --- a/backend/modules/fleet/views.py +++ b/backend/modules/fleet/views.py @@ -2450,6 +2450,26 @@ def _file_attachment(file_field, fallback_name): return (filename, content, _guess_content_type(filename)) +def _write_file_field_to_zip(archive, *, file_field, entry_name, chunk_size=64 * 1024): + if not file_field: + return False + file_field.open('rb') + try: + first_chunk = file_field.read(chunk_size) + if not first_chunk: + return False + with archive.open(entry_name, mode='w') as destination: + destination.write(first_chunk) + while True: + chunk = file_field.read(chunk_size) + if not chunk: + break + destination.write(chunk) + finally: + file_field.close() + return True + + def _build_image_attachments_for_work_order(work_order): attachments = [] photos = WorkOrderPhoto.objects.filter(is_active=True, work_order=work_order).order_by('created_at') @@ -3008,7 +3028,7 @@ def _parse_year_month_params(request): return year, month -def _build_monthly_service_tasks_archive_content(*, user, year, month): +def _write_monthly_service_tasks_archive_entries(archive, *, user, year, month): from modules.task_management.models import Task tasks = list( @@ -3028,24 +3048,37 @@ def _build_monthly_service_tasks_archive_content(*, user, year, month): raise DRFValidationError({'detail': 'Nema servisnih taskova za odabrani mjesec.'}) used_names = set() + entries_written = 0 + for task in tasks: + work_order = task.work_order + if work_order is None: + continue + docx_bytes = _build_work_order_service_records_docx_bytes(work_order, related_tasks=[task]) + base_name = _service_records_docx_filename(work_order, task) + entry_name = _unique_zip_entry_name(base_name, used_names) + archive.writestr(entry_name, docx_bytes) + entries_written += 1 + if entries_written == 0: + raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) + return entries_written + + +def _build_monthly_service_tasks_archive_content(*, user, year, month): archive_buffer = BytesIO() with zipfile.ZipFile(archive_buffer, mode='w', compression=zipfile.ZIP_DEFLATED) as archive: - for task in tasks: - work_order = task.work_order - if work_order is None: - continue - docx_bytes = _build_work_order_service_records_docx_bytes(work_order, related_tasks=[task]) - base_name = _service_records_docx_filename(work_order, task) - entry_name = _unique_zip_entry_name(base_name, used_names) - archive.writestr(entry_name, docx_bytes) - + _write_monthly_service_tasks_archive_entries( + archive, + user=user, + year=year, + month=month, + ) archive_content = archive_buffer.getvalue() if not archive_content: raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) return archive_content -def _build_monthly_work_orders_archive_content(*, user, year, month): +def _write_monthly_work_orders_archive_entries(archive, *, user, year, month): from modules.task_management.models import Task monthly_tasks = ( @@ -3084,23 +3117,34 @@ def _build_monthly_work_orders_archive_content(*, user, year, month): raise DRFValidationError({'detail': 'Nema putnih naloga za odabrani mjesec.'}) used_names = set() + entries_written = 0 + for work_order in work_orders: + pdf_bytes = _build_work_order_pdf(work_order) + work_order_pdf_name = _unique_zip_entry_name(_pdf_filename(work_order, 'work_order'), used_names) + archive.writestr(work_order_pdf_name, pdf_bytes) + entries_written += 1 + + display_code = _work_order_display_code(work_order) + invoices = work_order.invoices.filter(is_active=True).order_by('datum', 'created_at') + for index, invoice in enumerate(invoices, start=1): + file_name = Path(str(getattr(getattr(invoice, 'image', None), 'name', '') or f"invoice-{index}.bin")).name + archive_path = _unique_zip_entry_name(f"Racuni/{display_code}/{file_name}", used_names) + if _write_file_field_to_zip(archive, file_field=invoice.image, entry_name=archive_path): + entries_written += 1 + if entries_written == 0: + raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) + return entries_written + + +def _build_monthly_work_orders_archive_content(*, user, year, month): archive_buffer = BytesIO() with zipfile.ZipFile(archive_buffer, mode='w', compression=zipfile.ZIP_DEFLATED) as archive: - for work_order in work_orders: - pdf_bytes = _build_work_order_pdf(work_order) - work_order_pdf_name = _unique_zip_entry_name(_pdf_filename(work_order, 'work_order'), used_names) - archive.writestr(work_order_pdf_name, pdf_bytes) - - display_code = _work_order_display_code(work_order) - invoices = work_order.invoices.filter(is_active=True).order_by('datum', 'created_at') - for index, invoice in enumerate(invoices, start=1): - attachment = _file_attachment(invoice.image, fallback_name=f"invoice-{index}.bin") - if not attachment: - continue - invoice_filename, content, _content_type = attachment - archive_path = f"Racuni/{display_code}/{invoice_filename}" - archive.writestr(_unique_zip_entry_name(archive_path, used_names), content) - + _write_monthly_work_orders_archive_entries( + archive, + user=user, + year=year, + month=month, + ) archive_content = archive_buffer.getvalue() if not archive_content: raise DRFValidationError({'detail': 'Nije moguće kreirati ZIP za odabrani mjesec.'}) @@ -3411,12 +3455,18 @@ def generated_archive_download(request, archive_id): if generated_archive is None: raise DRFValidationError({'detail': 'ZIP arhiva nije dostupna ili je istekla.'}) - generated_archive.file.open('rb') + file_name = getattr(generated_archive.file, 'name', '') + storage = getattr(generated_archive.file, 'storage', None) + if not file_name or storage is None: + raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) try: - archive_bytes = generated_archive.file.read() - finally: - generated_archive.file.close() - if not archive_bytes: + if not storage.exists(file_name): + raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) + except (FileNotFoundError, OSError, ValueError): + raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) + try: + generated_archive.file.open('rb') + except (FileNotFoundError, OSError, ValueError): raise DRFValidationError({'detail': 'ZIP arhiva je prazna ili nedostupna.'}) filename = generated_archive.filename or _generated_archive_filename_for_user( request.user, @@ -3424,10 +3474,13 @@ def generated_archive_download(request, archive_id): month=generated_archive.month, archive_type=generated_archive.archive_type, ) - response = HttpResponse(archive_bytes, content_type='application/zip') + response = FileResponse(generated_archive.file, content_type='application/zip') response['Content-Disposition'] = f'attachment; filename="{filename}"' response['Cache-Control'] = 'private, max-age=3600' - response['Content-Length'] = str(len(archive_bytes)) + try: + response['Content-Length'] = str(generated_archive.file.size) + except (OSError, ValueError, TypeError): + pass return response diff --git a/backend/scripts/cleanup_tmp_files.sh b/backend/scripts/cleanup_tmp_files.sh new file mode 100644 index 0000000..e4c7be6 --- /dev/null +++ b/backend/scripts/cleanup_tmp_files.sh @@ -0,0 +1,24 @@ +#!/bin/sh +set -eu + +TMP_DIR="${APP_TMP_CLEANUP_DIR:-/tmp}" +MAX_AGE_DAYS="${APP_TMP_CLEANUP_MAX_AGE_DAYS:-1}" +PREFIXES="${APP_TMP_CLEANUP_PREFIXES:-erp-fleet-archive-}" + +if [ ! -d "$TMP_DIR" ]; then + echo "TMP dir ne postoji: $TMP_DIR" + exit 0 +fi + +OLD_IFS="$IFS" +IFS=',' +for prefix in $PREFIXES; do + prefix_trimmed="$(echo "$prefix" | xargs)" + if [ -z "$prefix_trimmed" ]; then + continue + fi + find "$TMP_DIR" -maxdepth 1 -mindepth 1 -name "${prefix_trimmed}*" -mtime "+${MAX_AGE_DAYS}" -print -delete +done +IFS="$OLD_IFS" + +echo "TMP cleanup dovršen za $TMP_DIR (older than ${MAX_AGE_DAYS} day(s))." diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 03e110e..30449f6 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -37,7 +37,13 @@ services: environment: DEBUG: "False" DJANGO_SETTINGS_MODULE: core.settings.production - command: celery -A core worker --loglevel=info + CELERY_WORKER_PREFETCH_MULTIPLIER: "1" + CELERY_WORKER_MAX_TASKS_PER_CHILD: "20" + CELERY_WORKER_MAX_MEMORY_PER_CHILD: "350000" + APP_TMP_CLEANUP_DIR: "/tmp" + APP_TMP_CLEANUP_MAX_AGE_HOURS: "24" + APP_TMP_CLEANUP_PREFIXES: "erp-fleet-archive-" + command: celery -A core worker --loglevel=info --concurrency=${CELERY_WORKER_CONCURRENCY:-2} volumes: - media_volume:/app/media depends_on: