diff --git a/backend/core/process/builders/upload_builder.py b/backend/core/process/builders/upload_builder.py index 19fce55..7952d53 100644 --- a/backend/core/process/builders/upload_builder.py +++ b/backend/core/process/builders/upload_builder.py @@ -28,7 +28,8 @@ def build(self): if not check.get("success"): raise ValueError(check.get("message")) - product.status = ProductStatus.PROCESSING + created_product = product.data + created_product.status = ProductStatus.PROCESSING product.save() - return product.data + return created_product diff --git a/backend/core/process/service.py b/backend/core/process/service.py index 57f1453..e5e9017 100644 --- a/backend/core/process/service.py +++ b/backend/core/process/service.py @@ -44,8 +44,11 @@ def submit(self): ) self.process.orchestration_process_id = orch_process.get("id") - self.process.used_config = json.loads(orch_process.get("used_config", "{}")) self.process.path = orch_process.get("path_str") + orch_used_config = orch_process.get("used_config", {}) + if isinstance(orch_used_config, str): + orch_used_config = json.loads(orch_used_config or "{}") + self.process.used_config = orch_used_config self.process.save() def stop(self): diff --git a/backend/core/test/test_process_logs.py b/backend/core/test/test_process_logs.py new file mode 100644 index 0000000..b33c4b6 --- /dev/null +++ b/backend/core/test/test_process_logs.py @@ -0,0 +1,28 @@ +import tempfile +from pathlib import Path + +from core.utils import get_logs +from django.test import SimpleTestCase, override_settings + + +class GetLogsTestCase(SimpleTestCase): + def test_returns_empty_logs_when_paths_are_missing(self): + logs = get_logs(None, None) + + self.assertEqual("", logs["pipeline"]["content"]) + self.assertEqual("", logs["slurm"]["content"]) + + def test_reads_logs_from_product_dir_when_process_dir_is_missing(self): + with tempfile.TemporaryDirectory() as upload_root: + product_relpath = "validation_results/sample-product" + pipeline_log = ( + Path(upload_root) / product_relpath / "process_info" / "pipeline.log" + ) + pipeline_log.parent.mkdir(parents=True, exist_ok=True) + pipeline_log.write_text("pipeline output", encoding="utf-8") + + with override_settings(UPLOAD_DIR=upload_root): + logs = get_logs(None, product_relpath) + + self.assertEqual("pipeline output", logs["pipeline"]["content"]) + self.assertEqual("", logs["slurm"]["content"]) diff --git a/backend/core/test/test_process_service.py b/backend/core/test/test_process_service.py new file mode 100644 index 0000000..9e30e4f --- /dev/null +++ b/backend/core/test/test_process_service.py @@ -0,0 +1,116 @@ +from types import SimpleNamespace +from unittest import mock + +from core.models import Pipeline, Process, Product, ProductStatus, ProductType +from core.process.builders.upload_builder import UploadBuilder +from core.process.service import ProcessService +from django.contrib.auth.models import User +from django.test import TestCase + + +class UploadBuilderTestCase(TestCase): + def setUp(self): + self.user = User.objects.create_user( + "john", "john@snow.com", "you_know_nothing" + ) + self.product_type = ProductType.objects.create( + name="validation_results", + display_name="Validation Results", + ) + self.pipeline = Pipeline.objects.create( + name="combine_redshift_dedup", + display_name="Combine Redshift Dedup", + version="1.0.0", + output_product_type=self.product_type, + ) + + def test_build_sets_upload_status_to_processing(self): + serializer = SimpleNamespace( + initial_data={ + "pipeline": self.pipeline.pk, + "display_name": "Combined Catalog", + "description": "Generated by process", + } + ) + + product = UploadBuilder(serializer, self.user).build() + + self.assertEqual(ProductStatus.PROCESSING, product.status) + self.assertIsNotNone(product.path) + + +class ProcessServiceTestCase(TestCase): + def setUp(self): + self.user = User.objects.create_user( + "john", "john@snow.com", "you_know_nothing" + ) + self.product_type = ProductType.objects.create( + name="redshift_catalog", + display_name="Redshift Catalog", + ) + self.pipeline = Pipeline.objects.create( + name="combine_redshift_dedup", + display_name="Combine Redshift Dedup", + version="1.0.0", + output_product_type=self.product_type, + ) + self.upload = Product.objects.create( + product_type=self.product_type, + user=self.user, + display_name="Combined Catalog", + internal_name="1_combined_catalog", + path="redshift_catalog/1_combined_catalog", + status=ProductStatus.PROCESSING, + ) + self.process = Process.objects.create( + display_name="Combined Catalog", + pipeline=self.pipeline, + upload=self.upload, + user=self.user, + ) + + @mock.patch("core.process.service.InputsBuilder.build", return_value=[]) + @mock.patch("core.process.service.BasePipelineHandler.get_handler") + @mock.patch("core.process.service.Maestro") + def test_submit_accepts_dict_used_config( + self, + maestro_cls, + get_handler, + _build_inputs, + ): + class FakeHandler: + def __init__(self, request, process): + self.request = request + self.process = process + + def build_config(self): + return {"param": {"debug": True}} + + get_handler.return_value = FakeHandler + + maestro = maestro_cls.return_value + maestro.start.return_value = { + "id": 123, + "path_str": "combine_redshift_dedup/00000123", + "used_config": { + "param": {"debug": True}, + "inputs": {"specz": []}, + "output_dir": self.upload.path, + }, + } + + request = SimpleNamespace(data={}) + + ProcessService(request, self.process).submit() + self.process.refresh_from_db() + + self.assertEqual(123, self.process.orchestration_process_id) + self.assertEqual("combine_redshift_dedup/00000123", self.process.path) + self.assertEqual( + { + "param": {"debug": True}, + "inputs": {"specz": []}, + "output_dir": self.upload.path, + }, + self.process.used_config, + ) diff --git a/backend/core/utils.py b/backend/core/utils.py index e985bff..85f3998 100644 --- a/backend/core/utils.py +++ b/backend/core/utils.py @@ -36,8 +36,17 @@ def get_logs(process_dir, product_dir): logs = get_lognames() - process_dir = pathlib.Path(settings.PROCESSING_DIR, process_dir) - product_dir = pathlib.Path(settings.UPLOAD_DIR, product_dir) + search_roots = [] + + if process_dir: + search_roots.append( + ("process_dir", pathlib.Path(settings.PROCESSING_DIR, process_dir)) + ) + + if product_dir: + search_roots.append( + ("product_dir", pathlib.Path(settings.UPLOAD_DIR, product_dir)) + ) for key, data in logs.items(): log = [] @@ -45,11 +54,11 @@ def get_logs(process_dir, product_dir): logger.debug(f"Searching for log files: {key}") for fname in lognames: - log = glob.glob(f"{process_dir}/**/{fname}", recursive=True) - logger.debug(f"process_dir: {process_dir}/**/{fname} -> {log}") - if not log: - log = glob.glob(f"{product_dir}/**/{fname}", recursive=True) - logger.debug(f"product_dir: {product_dir}/**/{fname} -> {log}") + for root_name, root_path in search_roots: + log = glob.glob(f"{root_path}/**/{fname}", recursive=True) + logger.debug(f"{root_name}: {root_path}/**/{fname} -> {log}") + if log: + break if log: break diff --git a/backend/core/views/process.py b/backend/core/views/process.py index e10dd6a..8cda303 100644 --- a/backend/core/views/process.py +++ b/backend/core/views/process.py @@ -2,7 +2,7 @@ import requests from core.maestro import Maestro -from core.models import Process +from core.models import Process, ProductStatus from core.process.builders.upload_builder import UploadBuilder from core.process.service import ProcessService from core.serializers import ProcessSerializer @@ -84,7 +84,19 @@ def create(self, request): try: ProcessService(request, process).submit() except Exception as err: - LOGGER.error(err) + process.status = "Failed" + process.comment = str(err) + process.save(update_fields=["status", "comment"]) + process.upload.status = ProductStatus.FAILED + process.upload.save() + LOGGER.exception( + "Process submission failed: process_id=%s upload_id=%s orchestration_process_id=%s process_path=%s upload_path=%s", + process.pk, + process.upload_id, + process.orchestration_process_id, + process.path, + process.upload.path, + ) return Response( {"error": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, diff --git a/backend/pzserver/test_settings.py b/backend/pzserver/test_settings.py index 06c68c5..7e03fda 100644 --- a/backend/pzserver/test_settings.py +++ b/backend/pzserver/test_settings.py @@ -127,8 +127,9 @@ STATIC_URL = "/django_static/" STATIC_ROOT = os.path.join(BASE_DIR, "django_static") +UPLOAD_DIR = os.getenv("UPLOAD_DIR", os.path.join(BASE_DIR, "archive/data/")) MEDIA_URL = os.path.join(BASE_DIR, "archive/data/") -MEDIA_ROOT = os.path.join(BASE_DIR, "archive/data/") +MEDIA_ROOT = os.getenv("MEDIA_ROOT", UPLOAD_DIR) PRODUCT_DOWNLOAD_COMPRESSION_LEVEL = int( os.getenv("PRODUCT_DOWNLOAD_COMPRESSION_LEVEL", "6") ) @@ -151,6 +152,8 @@ os.getenv("PRODUCT_DOWNLOAD_PREPARE_POLL_INTERVAL_SECONDS", "0.5") ) +PROCESSING_DIR = os.getenv("PROCESSING_DIR", os.path.join(BASE_DIR, "processes")) + # Default primary key field type # https://docs.djangoproject.com/en/4.0/ref/settings/#default-auto-field diff --git a/orchestration b/orchestration index bce68d4..dca674d 160000 --- a/orchestration +++ b/orchestration @@ -1 +1 @@ -Subproject commit bce68d43b13723bfb04a9ec73d96b38d5d1ed0ce +Subproject commit dca674d7b64d6df19371ae482dc601d819f6f980