diff --git a/gateway/scripts/diagnose_metadata_files.py b/gateway/scripts/diagnose_metadata_files.py new file mode 100755 index 00000000..3d636643 --- /dev/null +++ b/gateway/scripts/diagnose_metadata_files.py @@ -0,0 +1,221 @@ +#!/usr/bin/env python3 +# ruff: noqa: T201, BLE001, E402, C901 +"""Diagnose corrupt/unreadable DigitalRF metadata files for captures. + +Usage: + python scripts/diagnose_metadata_files.py + python scripts/diagnose_metadata_files.py [file_uuid ...] + python scripts/diagnose_metadata_files.py --file ids.txt + +For each File record, fetches the object from MinIO to a temp dir, +inspects with h5py, and reports findings. + +The --file mode reads UUIDs from a text file (one UUID per line, +blank lines and lines starting with '#' are ignored). +""" + +from __future__ import annotations + +import os +import sys +import tempfile +from pathlib import Path + +# Ensure the project root is on sys.path so the 'config' package is importable. +# Works whether run from project root (cd /app) or via absolute script path. +_PROJECT_ROOT = Path(__file__).resolve().parent.parent +if str(_PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(_PROJECT_ROOT)) + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.production") + +import django + +django.setup() + +import h5py +from django.conf import settings +from sds_gateway.api_methods.models import Capture +from sds_gateway.api_methods.models import File +from sds_gateway.api_methods.utils.minio_client import get_minio_client + + +def inspect_file(minio_client, bucket: str, file_obj: File) -> dict: + """Download a file from MinIO to a temp dir and inspect it.""" + result = { + "uuid": str(file_obj.uuid), + "name": file_obj.name, + "directory": file_obj.directory, + "s3_key": file_obj.file.name if file_obj.file else "NONE", + } + if not file_obj.file: + result["error"] = "No file field (S3 key) on File record" + return result + + with tempfile.TemporaryDirectory() as tmpdir: + local_path = Path(tmpdir) / file_obj.name + try: + minio_client.fget_object( + bucket_name=bucket, + object_name=file_obj.file.name, + file_path=str(local_path), + ) + except Exception as exc: + result["error"] = f"MinIO fetch failed: {type(exc).__name__}: {exc}" + return result + + file_size = local_path.stat().st_size + result["size_bytes"] = file_size + + if file_size == 0: + result["status"] = "EMPTY_FILE (0 bytes)" + return result + + # Try to open with h5py + try: + with h5py.File(str(local_path), "r") as f: + keys = list(f.keys()) + result["hdf5_groups"] = keys + result["hdf5_group_count"] = len(keys) + if keys: + result["first_key"] = keys[0] + result["last_key"] = keys[-1] + # Check attributes + attrs = dict(f.attrs) + result["hdf5_attrs"] = { + k: (str(v) if isinstance(v, bytes) else v) for k, v in attrs.items() + } + result["status"] = "OK" + except Exception as exc: + result["error"] = f"h5py open failed: {type(exc).__name__}: {exc}" + result["status"] = "CORRUPT_HDF5" + return result + + +def _read_ids_from_file(path: str) -> list[str]: + """Read UUIDs from a file, one per line. Skips blanks and comments.""" + ids: list[str] = [] + with Path(path).open() as fh: + for raw in fh: + stripped = raw.strip() + if not stripped or stripped.startswith("#"): + continue + ids.append(stripped) + return ids + + +def _files_for_id(target: str) -> list[File]: + """Resolve a UUID to File objects: try Capture first, then File.""" + files: list[File] = [] + + # Try as capture UUID first + try: + capture = Capture.objects.get(uuid=target, is_deleted=False) + files = list(capture.files.filter(is_deleted=False).order_by("name")) + print(f"\nCapture: {capture.uuid} ({capture.capture_type}, {capture.name})") + print(f"Channel: {capture.channel}") + print(f"Top-level dir: {capture.top_level_dir}") + print(f"Files connected: {len(files)}") + except Capture.DoesNotExist: + pass + else: + return files + + # Try as file UUID + try: + f = File.objects.get(uuid=target, is_deleted=False) + except File.DoesNotExist: + pass + else: + return [f] + + print(f"Not found: {target}") + return files + + +def _print_results(results: list[dict]) -> None: + """Print a summary of inspection results.""" + by_status: dict[str, list] = {} + for r in results: + status = r.get("status", "UNKNOWN") + by_status.setdefault(status, []).append(r) + + for status, items in sorted(by_status.items()): + print(f"\n[{status}] — {len(items)} file(s):") + for item in items: + print(f" {item['name']:40s} s3_key={item['s3_key']}") + if "size_bytes" in item: + print(f" size={item['size_bytes']} bytes") + if "error" in item: + print(f" ERROR: {item['error']}") + if "hdf5_groups" in item: + print( + f" groups={item['hdf5_group_count']}," + f" first={item.get('first_key')}," + f" last={item.get('last_key')}" + ) + if "hdf5_attrs" in item: + for k, v in item["hdf5_attrs"].items(): + print(f" attr {k}={v}") + + # Specific checks for dmd_properties.h5 + dmd_props = [r for r in results if "dmd_properties" in r["name"]] + if not dmd_props: + print("\n\u26a0\ufe0f WARNING: No dmd_properties.h5 file found!") + print(" This is required for DigitalMetadataReader initialization.") + + metadata_files = [r for r in results if r["name"].startswith("metadata@")] + if metadata_files: + print(f"\nMetadata data files ({len(metadata_files)}):") + for r in metadata_files: + status = r.get("status", "???") + print(f" {r['name']}: {status}") + + +def main() -> None: + min_cli_args = 2 + file_mode_args = 3 + + if len(sys.argv) < min_cli_args: + print(__doc__) + sys.exit(1) + + # Collect target IDs + targets: list[str] = [] + if sys.argv[1] == "--file": + if len(sys.argv) < file_mode_args: + print("Error: --file requires a path", file=sys.stderr) + sys.exit(2) + targets = _read_ids_from_file(sys.argv[2]) + if not targets: + print("No IDs found in file.", file=sys.stderr) + sys.exit(1) + else: + targets = sys.argv[1:] + + minio_client = get_minio_client() + bucket = settings.AWS_STORAGE_BUCKET_NAME + print(f"Bucket: {bucket}") + + all_results: list[dict] = [] + for target in targets: + files_to_check = _files_for_id(target) + if not files_to_check: + continue + + print(f"Checking {len(files_to_check)} file(s)...") + print("=" * 60) + + for fobj in files_to_check: + result = inspect_file(minio_client, bucket, fobj) + all_results.append(result) + + if not all_results: + print("No files inspected.") + sys.exit(1) + + _print_results(all_results) + + +if __name__ == "__main__": + main() diff --git a/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py b/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py index 39d08d86..a951b29c 100644 --- a/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py +++ b/gateway/sds_gateway/api_methods/helpers/extract_drf_metadata.py @@ -57,46 +57,67 @@ def read_metadata_by_channel( channel_name: str, ) -> dict[str, typing.Any]: """Reads Digital RF metadata.""" - - rf_reader = drf.DigitalRFReader(str(data_path)) - bounds_raw = rf_reader.get_bounds(channel_name) - bounds_raw = typing.cast("tuple[int, int]", bounds_raw) - bounds = Bounds(start=bounds_raw[0], end=bounds_raw[1]) - - # get properties - drf_properties = typing.cast( - "dict[str, typing.Any]", - rf_reader.get_properties(channel_name=channel_name, sample=bounds.start), - ) - - # add bounds to properties, convert to unix timestamp - drf_properties["start_bound"] = bounds.start / drf_properties["samples_per_second"] - drf_properties["end_bound"] = bounds.end / drf_properties["samples_per_second"] - - # initialize the digital metadata reader - dmd_properties = {} try: - md_reader = typing.cast( - "DigitalMetadataReader", - rf_reader.get_digital_metadata(channel_name), + rf_reader = drf.DigitalRFReader(str(data_path)) + bounds_raw = rf_reader.get_bounds(channel_name) + bounds_raw = typing.cast("tuple[int, int]", bounds_raw) + bounds = Bounds(start=bounds_raw[0], end=bounds_raw[1]) + + # get properties + drf_properties = typing.cast( + "dict[str, typing.Any]", + rf_reader.get_properties(channel_name=channel_name, sample=bounds.start), + ) + + # add bounds to properties, convert to unix timestamp + drf_properties["start_bound"] = ( + bounds.start / drf_properties["samples_per_second"] ) + drf_properties["end_bound"] = bounds.end / drf_properties["samples_per_second"] + + # initialize the digital metadata reader + dmd_properties = {} + try: + md_reader = typing.cast( + "DigitalMetadataReader", + rf_reader.get_digital_metadata(channel_name), + ) + except OSError as e: + msg = f"No digital metadata for channel '{channel_name}': {e}" + log.warning(msg) + else: + try: + dmd_properties = md_reader.read_flatdict( + start_sample=bounds.start, + method="ffill", + ) + if not isinstance(dmd_properties, dict): + msg = "Expected dmd_properties to be a dictionary" + raise TypeError(msg) + except OSError as e: + msg = ( + f"Digital metadata files for channel '{channel_name}' " + f"are missing, corrupt, or unreadable: {e}. " + f"Try re-uploading the metadata files for this channel, " + f"or contact support with your capture UUID." + ) + log.warning(msg) + raise ValueError(msg) from e except OSError as e: - msg = f"No digital metadata for channel '{channel_name}': {e}" + msg = ( + f"Digital RF data or metadata for channel '{channel_name}' " + f"is missing, corrupt, or unreadable: {e}. " + f"Try re-uploading the channel files, " + f"or contact support with this capture UUID." + ) log.warning(msg) + raise ValueError(msg) from e else: - dmd_properties = md_reader.read_flatdict( - start_sample=bounds.start, - method="ffill", - ) - if not isinstance(dmd_properties, dict): - msg = "Expected dmd_properties to be a dictionary" - raise TypeError(msg) - - # Merge the flattened dictionaries - return { - **drf_properties, - **dmd_properties, - } + # Merge the flattened dictionaries + return { + **drf_properties, + **dmd_properties, + } def validate_metadata_by_channel( diff --git a/gateway/sds_gateway/api_methods/helpers/reconstruct_file_tree.py b/gateway/sds_gateway/api_methods/helpers/reconstruct_file_tree.py index e7b173a2..bada8d5e 100644 --- a/gateway/sds_gateway/api_methods/helpers/reconstruct_file_tree.py +++ b/gateway/sds_gateway/api_methods/helpers/reconstruct_file_tree.py @@ -13,6 +13,8 @@ from sds_gateway.api_methods.utils.disk_utils import check_disk_space_available from sds_gateway.api_methods.utils.disk_utils import estimate_disk_size from sds_gateway.api_methods.utils.minio_client import get_minio_client +from sds_gateway.api_methods.utils.storage_errors import StorageUnavailableError +from sds_gateway.api_methods.utils.storage_errors import is_storage_unavailable_error from sds_gateway.users.models import User RH_DEFAULT_EXTENSION = ".rh.json" @@ -263,7 +265,11 @@ def _check_disk_space_available_for_reconstruction( # Check disk space only for files that will be fetched if files_to_fetch: estimated_size = estimate_disk_size(files_to_fetch) - if not check_disk_space_available(estimated_size, target_dir): + if not check_disk_space_available( + estimated_size, + target_dir, + buffer_bytes=0, + ): msg = ( f"Insufficient disk space for reconstructing file tree. " f"Required: {estimated_size} bytes, " @@ -366,11 +372,28 @@ def reconstruct_tree( filenames_of_interest=filenames_of_interest, ) if fetch_file_contents: - minio_client.fget_object( - bucket_name=settings.AWS_STORAGE_BUCKET_NAME, - object_name=file_obj.file.name, - file_path=str(local_file_path), - ) + try: + minio_client.fget_object( + bucket_name=settings.AWS_STORAGE_BUCKET_NAME, + object_name=file_obj.file.name, + file_path=str(local_file_path), + ) + except Exception as e: + if is_storage_unavailable_error(e): + msg = ( + f"Object storage is unavailable while fetching " + f"file '{file_obj.name}': {e}" + ) + log.exception(msg) + raise StorageUnavailableError(msg) from e + msg = ( + f"Failed to fetch file '{file_obj.name}' " + f"from storage: {e}. " + f"Try re-uploading the file, " + f"or contact support with this capture UUID." + ) + log.warning(msg) + raise ValueError(msg) from e else: local_file_path.touch() diff --git a/gateway/sds_gateway/api_methods/tests/test_capture_endpoints.py b/gateway/sds_gateway/api_methods/tests/test_capture_endpoints.py index 55cf1654..0cb567c0 100644 --- a/gateway/sds_gateway/api_methods/tests/test_capture_endpoints.py +++ b/gateway/sds_gateway/api_methods/tests/test_capture_endpoints.py @@ -14,6 +14,7 @@ from unittest.mock import MagicMock from unittest.mock import patch +import pytest from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.urls import reverse @@ -22,6 +23,9 @@ from rest_framework.test import APIClient from rest_framework.test import APITestCase +from sds_gateway.api_methods.helpers.extract_drf_metadata import ( + read_metadata_by_channel, +) from sds_gateway.api_methods.helpers.index_handling import index_capture_metadata from sds_gateway.api_methods.helpers.reconstruct_file_tree import ( _get_list_of_capture_files, @@ -42,6 +46,7 @@ from sds_gateway.api_methods.tests.factories import UserSharePermissionFactory from sds_gateway.api_methods.utils.metadata_schemas import get_mapping_by_capture_type from sds_gateway.api_methods.utils.opensearch_client import get_opensearch_client +from sds_gateway.api_methods.utils.storage_errors import StorageUnavailableError from sds_gateway.api_methods.views.capture_endpoints import _normalize_top_level_dir from sds_gateway.users.models import UserAPIKey from sds_gateway.visualizations.models import PostProcessedData @@ -2377,6 +2382,73 @@ def test_waterfall_slices_stream_success( assert stream_file.capture_id == capture.pk +class StorageUnavailableErrorTestCases(APITestCase): + """Test storage-unavailable handling in capture create/update endpoints.""" + + def setUp(self) -> None: + self.client = APIClient() + self.user = User.objects.create( + email="storage-error-user@example.com", + password="testpassword", # noqa: S106 + is_approved=True, + ) + _api_key, key = UserAPIKey.objects.create_key( + name="storage-error-key", + user=self.user, + ) + self.client.credentials(HTTP_AUTHORIZATION=f"Api-Key: {key}") + self.list_url = reverse("api:captures-list") + self.capture = Capture.objects.create( + owner=self.user, + capture_type=CaptureType.DigitalRF, + channel="ch0", + top_level_dir="/files/storage-error-user@example.com/test-dir", + index_name="captures-test-drf", + ) + self.detail_url = reverse( + "api:captures-detail", + kwargs={"pk": self.capture.uuid}, + ) + + @patch( + "sds_gateway.api_methods.views.capture_endpoints.reconstruct_tree", + ) + def test_create_capture_storage_unavailable_returns_503( + self, + mock_reconstruct_tree, + ) -> None: + mock_reconstruct_tree.side_effect = StorageUnavailableError( + "Object storage is unavailable", + ) + + response = self.client.post( + self.list_url, + data={ + "capture_type": CaptureType.DigitalRF, + "channel": "ch0", + "top_level_dir": "test-dir", + "index_name": "captures-test-drf", + }, + ) + + assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE + + @patch( + "sds_gateway.api_methods.views.capture_endpoints.reconstruct_tree", + ) + def test_update_capture_storage_unavailable_returns_503( + self, + mock_reconstruct_tree, + ) -> None: + mock_reconstruct_tree.side_effect = StorageUnavailableError( + "Object storage is unavailable", + ) + + response = self.client.put(self.detail_url) + + assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE + + class OpenSearchErrorTestCases(APITestCase): """Test cases for OpenSearch error handling in capture endpoints.""" @@ -2948,3 +3020,30 @@ def test_bulk_load_does_not_crash_on_mismatched_hits( # Should not raise - cap2 gets empty cache response = self.client.get(self.list_url) assert response.status_code == status.HTTP_200_OK + + +def test_read_metadata_by_channel_unreadable_metadata() -> None: + """read_metadata_by_channel raises ValueError when metadata@*.h5 files + are corrupt/unreadable. + + The outer get_digital_metadata() call succeeds (dmd_properties.h5 exists) + but the inner read_flatdict() call fails because the underlying + metadata@*.h5 data files are corrupt. The fix converts this OSError + into a ValueError so create() returns HTTP 400 instead of 500. + """ + mock_rf = MagicMock() + mock_rf.get_bounds.return_value = (0, 1000) + mock_rf.get_properties.return_value = {"samples_per_second": 1000} + mock_md = mock_rf.get_digital_metadata.return_value + mock_md.read_flatdict.side_effect = OSError( + "All attempts to read first sample failed", + ) + + with ( + patch( + "sds_gateway.api_methods.helpers.extract_drf_metadata.drf.DigitalRFReader", + return_value=mock_rf, + ), + pytest.raises(ValueError, match="corrupt, or unreadable"), + ): + read_metadata_by_channel(Path("/fake"), "ch0") diff --git a/gateway/sds_gateway/api_methods/tests/test_celery_tasks.py b/gateway/sds_gateway/api_methods/tests/test_celery_tasks.py index f922a8aa..a6c83c60 100644 --- a/gateway/sds_gateway/api_methods/tests/test_celery_tasks.py +++ b/gateway/sds_gateway/api_methods/tests/test_celery_tasks.py @@ -238,13 +238,14 @@ def test_send_dataset_files_email_success( # Clear any existing emails mail.outbox.clear() - # Test the task - result = send_item_files_email.delay( - str(self.dataset.uuid), str(self.user.id), ItemType.DATASET - ) - - # Get the result - task_result = result.get() + with patch( + "sds_gateway.api_methods.tasks.check_disk_space_available", + return_value=True, + ): + result = send_item_files_email.delay( + str(self.dataset.uuid), str(self.user.id), ItemType.DATASET + ) + task_result = result.get() # Verify task completed successfully assert task_result["status"] == "success", ( @@ -398,13 +399,14 @@ def test_send_dataset_files_email_download_failure( # Clear any existing emails mail.outbox.clear() - # Test the task - result = send_item_files_email.delay( - str(self.dataset.uuid), str(self.user.id), ItemType.DATASET - ) - - # Get the result - task_result = result.get() + with patch( + "sds_gateway.api_methods.tasks.check_disk_space_available", + return_value=True, + ): + result = send_item_files_email.delay( + str(self.dataset.uuid), str(self.user.id), ItemType.DATASET + ) + task_result = result.get() # Verify task completed with error assert task_result["status"] == "error", ( @@ -823,6 +825,10 @@ def test_check_disk_space_available(self, mock_disk_usage): result = check_disk_space_available(500 * 1024 * 1024 * 1024) # 500GB required assert result is False + # Small temp-dir operations can skip the safety buffer. + result = check_disk_space_available(1024, buffer_bytes=0) + assert result is True + # Test with error handling mock_disk_usage.side_effect = OSError("Disk error") result = check_disk_space_available(1024 * 1024 * 1024) diff --git a/gateway/sds_gateway/api_methods/tests/test_minio.py b/gateway/sds_gateway/api_methods/tests/test_minio.py index 62a4a1b0..fa1ca0fd 100644 --- a/gateway/sds_gateway/api_methods/tests/test_minio.py +++ b/gateway/sds_gateway/api_methods/tests/test_minio.py @@ -17,6 +17,7 @@ from sds_gateway.api_methods.models import CaptureType from sds_gateway.api_methods.serializers.file_serializers import FilePostSerializer from sds_gateway.api_methods.utils.minio_client import get_minio_client +from sds_gateway.api_methods.utils.storage_errors import StorageUnavailableError if TYPE_CHECKING: from sds_gateway.users.models import User as UserModel @@ -242,6 +243,74 @@ def test_reconstruct_tree_insufficient_disk_space( assert "Insufficient disk space" in error_message assert "Required: 1073741824 bytes" in error_message + @patch( + "sds_gateway.api_methods.helpers.reconstruct_file_tree.check_disk_space_available" + ) + @patch("sds_gateway.api_methods.helpers.reconstruct_file_tree.estimate_disk_size") + @patch("sds_gateway.api_methods.helpers.reconstruct_file_tree.get_minio_client") + def test_reconstruct_tree_storage_unavailable( + self, + mock_get_minio_client, + mock_estimate_size, + mock_check_space, + ) -> None: + """Storage connection failures should not be converted to ValueError.""" + mock_check_space.return_value = True + mock_estimate_size.return_value = 1024 + mock_get_minio_client.return_value.fget_object.side_effect = ( + ConnectionRefusedError("Connection refused") + ) + + with ( + tempfile.TemporaryDirectory() as temp_dir, + pytest.raises( + StorageUnavailableError, match="Object storage is unavailable" + ), + ): + reconstruct_tree( + target_dir=Path(temp_dir), + virtual_top_dir=self.top_level_dir, + owner=self.user, + capture_type=CaptureType.RadioHound, + rh_scan_group=self.scan_group, + verbose=True, + ) + + @patch( + "sds_gateway.api_methods.helpers.reconstruct_file_tree.check_disk_space_available" + ) + @patch("sds_gateway.api_methods.helpers.reconstruct_file_tree.estimate_disk_size") + @patch("sds_gateway.api_methods.helpers.reconstruct_file_tree.get_minio_client") + def test_reconstruct_tree_missing_object_raises_value_error( + self, + mock_get_minio_client, + mock_estimate_size, + mock_check_space, + ) -> None: + """Missing storage objects should remain user-actionable ValueErrors.""" + + class MissingObjectError(Exception): + code = "NoSuchKey" + + mock_check_space.return_value = True + mock_estimate_size.return_value = 1024 + mock_get_minio_client.return_value.fget_object.side_effect = MissingObjectError( + "missing", + ) + + with ( + tempfile.TemporaryDirectory() as temp_dir, + pytest.raises(ValueError, match="Failed to fetch file"), + ): + reconstruct_tree( + target_dir=Path(temp_dir), + virtual_top_dir=self.top_level_dir, + owner=self.user, + capture_type=CaptureType.RadioHound, + rh_scan_group=self.scan_group, + verbose=True, + ) + class ReconstructDRFFileTreeTest(APITestCase): """TODO: Test reconstructing Digital RF file trees.""" diff --git a/gateway/sds_gateway/api_methods/tests/test_storage_errors.py b/gateway/sds_gateway/api_methods/tests/test_storage_errors.py new file mode 100644 index 00000000..077e1a50 --- /dev/null +++ b/gateway/sds_gateway/api_methods/tests/test_storage_errors.py @@ -0,0 +1,59 @@ +"""Tests for object-store error classification helpers.""" + +import pytest + +from sds_gateway.api_methods.utils.storage_errors import StorageUnavailableError +from sds_gateway.api_methods.utils.storage_errors import is_missing_object_error +from sds_gateway.api_methods.utils.storage_errors import is_storage_unavailable_error + + +class MissingObjectError(Exception): + """Test-only exception to simulate missing-object failures.""" + + code = "NoSuchKey" + + +class ServiceUnavailableObjectError(Exception): + """Test-only exception to simulate storage service failures.""" + + code = "ServiceUnavailable" + + +def test_is_missing_object_error_detects_no_such_key() -> None: + assert is_missing_object_error(MissingObjectError("missing")) + + +def test_is_storage_unavailable_error_detects_connection_refused() -> None: + assert is_storage_unavailable_error(ConnectionRefusedError("refused")) + + +def test_is_storage_unavailable_error_detects_timeout() -> None: + assert is_storage_unavailable_error(TimeoutError("timed out")) + + +def test_is_storage_unavailable_error_detects_urllib3_max_retry() -> None: + max_retry_error = type( + "MaxRetryError", + (Exception,), + {"__module__": "urllib3.exceptions"}, + )("retries exceeded") + + assert is_storage_unavailable_error(max_retry_error) + + +def test_is_storage_unavailable_error_does_not_classify_missing_object() -> None: + assert not is_storage_unavailable_error(MissingObjectError("missing")) + + +def test_is_storage_unavailable_error_detects_service_unavailable_code() -> None: + assert is_storage_unavailable_error( + ServiceUnavailableObjectError("service unavailable"), + ) + + +def test_storage_unavailable_error_is_os_error() -> None: + error = StorageUnavailableError("storage down") + + assert isinstance(error, OSError) + with pytest.raises(StorageUnavailableError): + raise error diff --git a/gateway/sds_gateway/api_methods/utils/disk_utils.py b/gateway/sds_gateway/api_methods/utils/disk_utils.py index 05ef169b..27fe6291 100644 --- a/gateway/sds_gateway/api_methods/utils/disk_utils.py +++ b/gateway/sds_gateway/api_methods/utils/disk_utils.py @@ -13,7 +13,10 @@ def check_disk_space_available( - required_bytes: int, directory: Path | None = None + required_bytes: int, + directory: Path | None = None, + *, + buffer_bytes: int | None = None, ) -> bool: """ Check if there's enough disk space available for the required bytes. @@ -21,6 +24,8 @@ def check_disk_space_available( Args: required_bytes: Number of bytes needed directory: Directory to check space for (defaults to MEDIA_ROOT) + buffer_bytes: Safety margin to reserve; defaults to DISK_SPACE_BUFFER. + Pass 0 for small temp-dir operations where only the payload size matters. Returns: bool: True if enough space is available, False otherwise @@ -28,9 +33,12 @@ def check_disk_space_available( if directory is None: directory = Path(settings.MEDIA_ROOT) + if buffer_bytes is None: + buffer_bytes = DISK_SPACE_BUFFER + try: _total, _used, free = shutil.disk_usage(directory) - available_space = free - DISK_SPACE_BUFFER + available_space = free - buffer_bytes except (OSError, ValueError) as e: logger.error(f"Error checking disk space for {directory}: {e}") return False diff --git a/gateway/sds_gateway/api_methods/utils/storage_errors.py b/gateway/sds_gateway/api_methods/utils/storage_errors.py index 9e6a2007..d7feb61b 100644 --- a/gateway/sds_gateway/api_methods/utils/storage_errors.py +++ b/gateway/sds_gateway/api_methods/utils/storage_errors.py @@ -1,5 +1,6 @@ """Shared predicates for object-store error classification.""" +from collections.abc import Iterator from typing import Final MISSING_OBJECT_ERROR_CODES: Final[set[str]] = { @@ -11,6 +12,41 @@ "NotFound", } +STORAGE_UNAVAILABLE_ERROR_CODES: Final[set[str]] = { + "InternalError", + "ServiceUnavailable", + "SlowDown", +} + +STORAGE_UNAVAILABLE_HTTP_STATUS_CODES: Final[set[str]] = { + "500", + "502", + "503", + "504", +} + +STORAGE_UNAVAILABLE_EXCEPTION_NAMES: Final[set[str]] = { + "ConnectTimeoutError", + "HostChangedError", + "MaxRetryError", + "NewConnectionError", + "ReadTimeoutError", +} + + +class StorageUnavailableError(OSError): + """Raised when object storage is unreachable or otherwise unavailable.""" + + +def _iter_exception_chain(error: BaseException) -> Iterator[BaseException]: + """Yield *error* and linked causes/contexts without cycles.""" + seen: set[int] = set() + current: BaseException | None = error + while current is not None and id(current) not in seen: + seen.add(id(current)) + yield current + current = current.__cause__ or current.__context__ + def is_missing_object_error(error: Exception) -> bool: """Return True when error represents a missing object/bucket condition. @@ -18,15 +54,47 @@ def is_missing_object_error(error: Exception) -> bool: Handles both MinIO SDK exceptions (``error.code``, ``error.status``) and botocore exceptions (``error.response["Error"]["Code"]``). """ - error_code = str(getattr(error, "code", "")) - if error_code in MISSING_OBJECT_ERROR_CODES: - return True - - response = getattr(error, "response", None) - if isinstance(response, dict): - code = str(response.get("Error", {}).get("Code", "")) - if code in MISSING_OBJECT_ERROR_CODES: + for current in _iter_exception_chain(error): + error_code = str(getattr(current, "code", "")) + if error_code in MISSING_OBJECT_ERROR_CODES: + return True + + response = getattr(current, "response", None) + if isinstance(response, dict): + code = str(response.get("Error", {}).get("Code", "")) + if code in MISSING_OBJECT_ERROR_CODES: + return True + + status_code = str(getattr(current, "status", "")) + if status_code == "404": + return True + + return False + + +def is_storage_unavailable_error(error: Exception) -> bool: + """Return True when error indicates object storage is unreachable.""" + if is_missing_object_error(error): + return False + + for current in _iter_exception_chain(error): + if isinstance(current, (ConnectionError, TimeoutError)): + return True + + exception_module = type(current).__module__ + exception_name = type(current).__name__ + if ( + exception_module.startswith("urllib3") + and exception_name in STORAGE_UNAVAILABLE_EXCEPTION_NAMES + ): + return True + + error_code = str(getattr(current, "code", "")) + if error_code in STORAGE_UNAVAILABLE_ERROR_CODES: + return True + + status_code = str(getattr(current, "status", "")) + if status_code in STORAGE_UNAVAILABLE_HTTP_STATUS_CODES: return True - status_code = str(getattr(error, "status", "")) - return status_code == "404" + return False diff --git a/gateway/sds_gateway/api_methods/views/capture_endpoints.py b/gateway/sds_gateway/api_methods/views/capture_endpoints.py index 94f23808..f729882a 100644 --- a/gateway/sds_gateway/api_methods/views/capture_endpoints.py +++ b/gateway/sds_gateway/api_methods/views/capture_endpoints.py @@ -68,6 +68,7 @@ detach_item_from_all_datasets, ) from sds_gateway.api_methods.utils.relationship_utils import get_capture_files +from sds_gateway.api_methods.utils.storage_errors import StorageUnavailableError from sds_gateway.api_methods.views.file_endpoints import sanitize_path_rel_to_user from sds_gateway.users.models import User from sds_gateway.visualizations.models import PostProcessedData @@ -529,6 +530,9 @@ def _handle_capture_creation_errors( if isinstance(error, ValueError): user_msg = f"Error handling metadata for capture '{capture.uuid}': {error}" return Response({"detail": user_msg}, status=status.HTTP_400_BAD_REQUEST) + if isinstance(error, StorageUnavailableError): + log.error("Object storage unavailable during capture creation: %s", error) + return Response(status=status.HTTP_503_SERVICE_UNAVAILABLE) if isinstance(error, os_exceptions.ConnectionError): user_msg = f"Error connecting to OpenSearch: {error}" log.error(user_msg) @@ -604,7 +608,12 @@ def create(self, request: Request) -> Response: get_serializer = CaptureGetSerializer(capture) return Response(get_serializer.data, status=status.HTTP_201_CREATED) - except (UnknownIndexError, ValueError, os_exceptions.ConnectionError) as e: + except ( + UnknownIndexError, + ValueError, + StorageUnavailableError, + os_exceptions.ConnectionError, + ) as e: # Transaction will auto-rollback, no manual deletion needed if isinstance(capture, Capture): return self._handle_capture_creation_errors(capture=capture, error=e) @@ -928,9 +937,11 @@ def update(self, request: Request, pk: str | None = None) -> Response: except ValueError as e: msg = f"Error handling metadata for capture '{target_capture.uuid}': {e}" return Response({"detail": msg}, status=status.HTTP_400_BAD_REQUEST) - except os_exceptions.ConnectionError as e: - msg = f"Error connecting to OpenSearch: {e}" - log.error(msg) + except (StorageUnavailableError, os_exceptions.ConnectionError) as e: + if isinstance(e, StorageUnavailableError): + log.error("Object storage unavailable during capture update: %s", e) + else: + log.error("Error connecting to OpenSearch: %s", e) return Response(status=status.HTTP_503_SERVICE_UNAVAILABLE) # Return updated capture