diff --git a/VERSION b/VERSION index cc6c9a4..e75da3e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.3.5 +2.3.6 diff --git a/log_manager/file_errors.py b/log_manager/file_errors.py new file mode 100644 index 0000000..9987819 --- /dev/null +++ b/log_manager/file_errors.py @@ -0,0 +1,37 @@ +import gzip +import hashlib +import zlib + + +FILE_READ_ERROR_CODE = "file_read_error" +FILE_READ_EXCEPTIONS = (EOFError, OSError, zlib.error) + + +def build_file_read_error(exc, stage): + return { + "code": FILE_READ_ERROR_CODE, + "kind": _get_error_kind(exc), + "stage": stage, + "exception": exc.__class__.__name__, + "message": str(exc), + } + + +def build_catalog_error_hash(collection_code, path): + identity = f"catalog-error\0{collection_code}\0{path}".encode("utf-8") + return hashlib.md5(identity).hexdigest() + + +def get_file_read_error(validation): + file_error = (validation or {}).get("file_error") or {} + if file_error.get("code") == FILE_READ_ERROR_CODE: + return file_error + return None + + +def _get_error_kind(exc): + if isinstance(exc, EOFError): + return "truncated" + if isinstance(exc, (gzip.BadGzipFile, zlib.error)): + return "corrupted" + return "io" diff --git a/log_manager/models.py b/log_manager/models.py index c6d9895..b3809cd 100644 --- a/log_manager/models.py +++ b/log_manager/models.py @@ -9,7 +9,7 @@ from collection.models import Collection from core.utils.date_utils import get_date_obj -from log_manager import choices +from log_manager import choices, file_errors class LogFile(models.Model): @@ -123,6 +123,8 @@ def for_collection_date(cls, collection, access_date, status_filters=None): if status_filters: queryset = queryset.filter(status__in=status_filters) + queryset = _exclude_file_read_errors(queryset) + return list(queryset) @classmethod @@ -146,14 +148,14 @@ def distinct_access_dates_for_parsing( status_filters, skip_hashes=None, ): + date_queryset = cls.objects.filter( + status__in=status_filters, + collection=collection, + date__gte=from_date, + date__lte=until_date, + ).exclude(hash__in=skip_hashes or []) date_queryset = ( - cls.objects.filter( - status__in=status_filters, - collection=collection, - date__gte=from_date, - date__lte=until_date, - ) - .exclude(hash__in=skip_hashes or []) + _exclude_file_read_errors(date_queryset) .values_list("date", flat=True) .distinct() .order_by("date") @@ -168,3 +170,11 @@ def distinct_access_dates_for_parsing( def __str__(self): return f"{self.path}" + + +def _exclude_file_read_errors(queryset): + read_error_ids = LogFile.objects.filter( + status=choices.LOG_FILE_STATUS_ERROR, + validation__file_error__code=file_errors.FILE_READ_ERROR_CODE, + ).values_list("pk", flat=True) + return queryset.exclude(pk__in=read_error_ids) diff --git a/log_manager/services/catalog.py b/log_manager/services/catalog.py index fad59b3..100c500 100644 --- a/log_manager/services/catalog.py +++ b/log_manager/services/catalog.py @@ -2,10 +2,12 @@ import os from django.conf import settings +from django.db import transaction +from django.utils import timezone from collection.models import Collection from core.utils import date_utils -from log_manager import models, utils +from log_manager import choices, file_errors, models, utils from log_manager_config import models as lmc_models @@ -51,6 +53,8 @@ def _catalog_log_files_in_directory( visible_dates, supported_extensions, ): + retry_paths = _get_file_read_error_paths(collection, directory_path) + for root, _sub_dirs, files in os.walk(directory_path): for name in files: _name, extension = os.path.splitext(name) @@ -58,23 +62,145 @@ def _catalog_log_files_in_directory( continue file_path = os.path.join(root, name) - file_stat = os.stat(file_path) + try: + file_stat = os.stat(file_path) + except file_errors.FILE_READ_EXCEPTIONS as exc: + logging.error( + "Error reading file metadata %s. Error: %s", + file_path, + exc, + ) + _record_file_read_error( + collection=collection, + path=file_path, + stat_result={}, + exc=exc, + ) + continue + file_ctime = date_utils.get_date_obj_from_timestamp(file_stat.st_ctime) logging.debug("Checking file %s with ctime %s.", file_path, file_ctime) - if file_ctime not in visible_dates: + if file_ctime not in visible_dates and file_path not in retry_paths: continue try: - models.LogFile.create_or_update( - collection=collection, - path=file_path, - stat_result=file_stat, - hash=utils.hash_file(file_path), - ) - except Exception as exc: + file_hash = utils.hash_file(file_path) + except file_errors.FILE_READ_EXCEPTIONS as exc: logging.error( "Error cataloging file %s. Error: %s", file_path, exc, ) + _record_file_read_error( + collection=collection, + path=file_path, + stat_result=file_stat, + exc=exc, + ) + continue + + _catalog_readable_file( + collection=collection, + path=file_path, + stat_result=file_stat, + file_hash=file_hash, + ) + + +def _get_file_read_error_paths(collection, directory_path): + return set( + models.LogFile.objects.filter( + collection=collection, + path__startswith=directory_path, + status=choices.LOG_FILE_STATUS_ERROR, + validation__file_error__code=file_errors.FILE_READ_ERROR_CODE, + ).values_list("path", flat=True) + ) + + +def _record_file_read_error(collection, path, stat_result, exc): + error_hash = file_errors.build_catalog_error_hash(collection.acron3, path) + with transaction.atomic(): + log_file = ( + models.LogFile.objects.select_for_update() + .filter( + collection=collection, + path=path, + status=choices.LOG_FILE_STATUS_ERROR, + validation__file_error__code=file_errors.FILE_READ_ERROR_CODE, + ) + .first() + ) + if log_file is None: + log_file = models.LogFile.create_or_update( + collection=collection, + path=path, + stat_result=stat_result, + hash=error_hash, + status=choices.LOG_FILE_STATUS_ERROR, + ) + + log_file.path = path + log_file.stat_result = stat_result + log_file.status = choices.LOG_FILE_STATUS_ERROR + log_file.date = None + log_file.validation = { + "file_error": file_errors.build_file_read_error(exc, stage="catalog") + } + log_file.summary = {} + log_file.last_processed_line = 0 + log_file.parse_heartbeat_at = None + log_file.save() + + +def _catalog_readable_file(collection, path, stat_result, file_hash): + with transaction.atomic(): + path_error = ( + models.LogFile.objects.select_for_update() + .filter( + collection=collection, + path=path, + status=choices.LOG_FILE_STATUS_ERROR, + validation__file_error__code=file_errors.FILE_READ_ERROR_CODE, + ) + .first() + ) + canonical = ( + models.LogFile.objects.select_for_update().filter(hash=file_hash).first() + ) + + if canonical and path_error and canonical.pk != path_error.pk: + path_error.delete() + canonical.updated = timezone.now() + canonical.save(update_fields=["updated"]) + return canonical + + log_file = canonical or path_error + if log_file: + if file_errors.get_file_read_error(log_file.validation): + _recover_readable_log_file(log_file, file_hash, path, stat_result) + else: + log_file.updated = timezone.now() + log_file.save(update_fields=["updated"]) + return log_file + + return models.LogFile.create_or_update( + collection=collection, + path=path, + stat_result=stat_result, + hash=file_hash, + ) + + +def _recover_readable_log_file(log_file, file_hash, path, stat_result): + log_file.hash = file_hash + log_file.path = path + log_file.stat_result = stat_result + log_file.status = choices.LOG_FILE_STATUS_CREATED + log_file.date = None + log_file.validation = {} + log_file.summary = {} + log_file.last_processed_line = 0 + log_file.parse_heartbeat_at = None + log_file.save() diff --git a/log_manager/services/validation.py b/log_manager/services/validation.py index 777ac47..d923bfd 100644 --- a/log_manager/services/validation.py +++ b/log_manager/services/validation.py @@ -2,7 +2,7 @@ from collection.models import Collection from core.utils import date_utils -from log_manager import choices, models, utils +from log_manager import choices, file_errors, models, utils from log_manager_config import models as lmc_models LOGFILE_STAT_RESULT_CTIME_INDEX = 9 @@ -184,7 +184,15 @@ def _update_log_file_with_validation_result( log_file.validation = validation_result log_file.validation.update({"buffer_size": buffer_size, "sample_size": sample_size}) - if validation_result.get("is_valid", {}).get("all", False): + content_error = validation_result.get("content", {}).get("error") or {} + if content_error.get("code") == file_errors.FILE_READ_ERROR_CODE: + log_file.validation["file_error"] = { + **content_error, + "stage": "validation", + } + log_file.date = None + log_file.status = choices.LOG_FILE_STATUS_ERROR + elif validation_result.get("is_valid", {}).get("all", False): log_file.date = validation_result.get("probably_date") or None log_file.status = choices.LOG_FILE_STATUS_QUEUED else: diff --git a/log_manager/tests/test_catalog.py b/log_manager/tests/test_catalog.py new file mode 100644 index 0000000..3219969 --- /dev/null +++ b/log_manager/tests/test_catalog.py @@ -0,0 +1,120 @@ +import gzip +import tempfile +from datetime import date +from pathlib import Path +from unittest.mock import patch + +from django.test import TestCase + +from collection.models import Collection +from log_manager import choices, file_errors, utils +from log_manager.models import LogFile +from log_manager.services import catalog + + +class CatalogLogFilesTests(TestCase): + def setUp(self): + self.collection = Collection.objects.create(acron3="per", acron2="pe") + + def _catalog_directory(self, directory, visible_dates=None): + catalog._catalog_log_files_in_directory( + collection=self.collection, + directory_path=directory, + visible_dates=visible_dates + if visible_dates is not None + else [date.today()], + supported_extensions=[".gz"], + ) + + def test_corrupted_gzip_is_recorded_once_as_error(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "2026-09-04_scielo.pe.log.gz" + path.write_bytes(b"\x1f\x8bcorrupted") + + self._catalog_directory(directory) + self._catalog_directory(directory) + + log_file = LogFile.objects.get() + expected_hash = file_errors.build_catalog_error_hash("per", str(path)) + + self.assertEqual(log_file.status, choices.LOG_FILE_STATUS_ERROR) + self.assertIsNone(log_file.date) + self.assertEqual(log_file.hash, expected_hash) + self.assertEqual( + log_file.validation["file_error"]["code"], + file_errors.FILE_READ_ERROR_CODE, + ) + self.assertEqual(log_file.validation["file_error"]["stage"], "catalog") + + def test_file_metadata_io_error_is_recorded(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "2026-09-04_scielo.pe.log.gz" + path.touch() + + with patch.object( + catalog.os, "stat", side_effect=PermissionError("denied") + ): + self._catalog_directory(directory) + + log_file = LogFile.objects.get() + + self.assertEqual(log_file.status, choices.LOG_FILE_STATUS_ERROR) + self.assertEqual(log_file.stat_result, {}) + self.assertEqual(log_file.validation["file_error"]["kind"], "io") + + def test_valid_replacement_recovers_the_error_record(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "2026-09-04_scielo.pe.log.gz" + path.write_bytes(b"\x1f\x8bcorrupted") + self._catalog_directory(directory) + log_file_id = LogFile.objects.get().pk + + with gzip.open(path, "wt") as output: + output.write( + "127.0.0.1 - - [04/Sep/2026:10:00:00 +0000] " + '"GET / HTTP/1.1" 200 1\n' + ) + expected_hash = utils.hash_file(path) + self._catalog_directory(directory) + + log_file = LogFile.objects.get() + + self.assertEqual(log_file.pk, log_file_id) + self.assertEqual(log_file.hash, expected_hash) + self.assertEqual(log_file.status, choices.LOG_FILE_STATUS_CREATED) + self.assertEqual(log_file.validation, {}) + + def test_read_error_is_retried_outside_the_visible_date_range(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "2026-09-04_scielo.pe.log.gz" + path.write_bytes(b"\x1f\x8bcorrupted") + self._catalog_directory(directory) + log_file_id = LogFile.objects.get().pk + + with gzip.open(path, "wb") as output: + output.write(b"repaired content\n") + self._catalog_directory(directory, visible_dates=[]) + + log_file = LogFile.objects.get() + + self.assertEqual(log_file.pk, log_file_id) + self.assertEqual(log_file.status, choices.LOG_FILE_STATUS_CREATED) + + def test_valid_duplicate_removes_the_error_placeholder(self): + with tempfile.TemporaryDirectory() as directory: + valid_path = Path(directory) / "2026-09-03_scielo.pe.log.gz" + repaired_path = Path(directory) / "2026-09-04_scielo.pe.log.gz" + content = b"same content\n" + with gzip.open(valid_path, "wb") as output: + output.write(content) + repaired_path.write_bytes(b"\x1f\x8bcorrupted") + self._catalog_directory(directory) + + with gzip.open(repaired_path, "wb") as output: + output.write(content) + self._catalog_directory(directory) + + expected_hash = utils.hash_file(valid_path) + + self.assertEqual(LogFile.objects.count(), 1) + self.assertEqual(LogFile.objects.get().hash, expected_hash) diff --git a/log_manager/tests/test_models.py b/log_manager/tests/test_models.py index 85eada8..d74ae86 100644 --- a/log_manager/tests/test_models.py +++ b/log_manager/tests/test_models.py @@ -1,3 +1,4 @@ +from datetime import date from unittest.mock import patch from django.db import IntegrityError @@ -42,3 +43,44 @@ def test_create_or_update_refetches_existing_after_integrity_error(self): ) self.assertEqual(log_file.pk, existing.pk) + + def test_parsing_candidates_exclude_file_read_errors(self): + read_error = LogFile.objects.create( + collection=self.collection, + path="/tmp/read-error.log.gz", + stat_result={}, + hash="2" * 32, + status=choices.LOG_FILE_STATUS_ERROR, + date=date(2026, 9, 4), + validation={ + "file_error": { + "code": "file_read_error", + "kind": "corrupted", + "stage": "validation", + } + }, + ) + parse_error = LogFile.objects.create( + collection=self.collection, + path="/tmp/parse-error.log.gz", + stat_result={}, + hash="3" * 32, + status=choices.LOG_FILE_STATUS_ERROR, + date=date(2026, 9, 4), + ) + + access_dates = LogFile.distinct_access_dates_for_parsing( + collection=self.collection, + from_date=date(2026, 9, 4), + until_date=date(2026, 9, 4), + status_filters=[choices.LOG_FILE_STATUS_ERROR], + ) + log_files = LogFile.for_collection_date( + collection=self.collection, + access_date=date(2026, 9, 4), + status_filters=[choices.LOG_FILE_STATUS_ERROR], + ) + + self.assertEqual(access_dates, [date(2026, 9, 4)]) + self.assertEqual(log_files, [parse_error]) + self.assertNotIn(read_error, log_files) diff --git a/log_manager/tests/test_validation.py b/log_manager/tests/test_validation.py index 957faf0..9f50747 100644 --- a/log_manager/tests/test_validation.py +++ b/log_manager/tests/test_validation.py @@ -1,3 +1,4 @@ +import gzip import tempfile from datetime import date from unittest.mock import patch @@ -51,6 +52,22 @@ def test_validate_file_returns_invalid_result_for_empty_log( ) self.assertIsNone(result["probably_date"]) + def test_validate_file_returns_structured_error_for_corrupted_gzip(self): + with tempfile.NamedTemporaryFile(suffix=".log.gz") as tmp_file: + compressed = bytearray(gzip.compress(b"valid line\n")) + compressed[-1] ^= 0xFF + tmp_file.write(compressed) + tmp_file.flush() + + result = utils.validate_file( + tmp_file.name, + sample_size=1.0, + buffer_size=2048, + ) + + self.assertEqual(result["content"]["error"]["code"], "file_read_error") + self.assertEqual(result["content"]["error"]["kind"], "corrupted") + @patch("log_manager.services.validation.utils.validate_file") def test_validate_log_file_updates_status_and_normalizes_result( self, mock_validate_file @@ -79,3 +96,55 @@ def test_validate_log_file_updates_status_and_normalizes_result( self.assertEqual(log_file.date, date(2026, 5, 10)) self.assertNotIn("datetimes", log_file.validation["content"]["summary"]) self.assertEqual(log_file.validation["probably_date"], "2026-05-10") + + @patch("log_manager.services.validation.utils.validate_file") + def test_validate_log_file_marks_read_failure_as_error(self, mock_validate_file): + log_file = LogFile.objects.create( + collection=self.collection, + path="/tmp/2026-05-10_access.log.gz", + stat_result={"size": 10}, + hash="3" * 32, + status=choices.LOG_FILE_STATUS_QUEUED, + date=date(2026, 5, 10), + ) + mock_validate_file.return_value = { + "probably_date": None, + "is_valid": {"all": False}, + "content": { + "summary": {"total_lines": {"error": "File is corrupted"}}, + "error": { + "code": "file_read_error", + "kind": "corrupted", + "message": "File is corrupted", + }, + }, + } + + validation.validate_log_file_and_update_status(log_file.hash) + + log_file.refresh_from_db() + self.assertEqual(log_file.status, choices.LOG_FILE_STATUS_ERROR) + self.assertIsNone(log_file.date) + self.assertEqual(log_file.validation["file_error"]["stage"], "validation") + + @patch("log_manager.services.validation.utils.validate_file") + def test_validate_log_file_keeps_readable_invalid_file_invalidated( + self, mock_validate_file + ): + log_file = LogFile.objects.create( + collection=self.collection, + path="/tmp/2026-05-10-error_access.log.gz", + stat_result={"size": 10}, + hash="4" * 32, + status=choices.LOG_FILE_STATUS_CREATED, + ) + mock_validate_file.return_value = { + "probably_date": date(2026, 5, 10), + "is_valid": {"all": False}, + "content": {"summary": {"total_lines": 10}}, + } + + validation.validate_log_file_and_update_status(log_file.hash) + + log_file.refresh_from_db() + self.assertEqual(log_file.status, choices.LOG_FILE_STATUS_INVALIDATED) diff --git a/log_manager/utils.py b/log_manager/utils.py index 16a996f..7ddf6f1 100644 --- a/log_manager/utils.py +++ b/log_manager/utils.py @@ -98,7 +98,7 @@ def _safe_sample_size(path, sample_size, buffer_size): try: total_lines = validator.get_total_lines(path=path, buffer_size=buffer_size) except ( - exceptions.TruncatedLogFileError, + exceptions.LogFileReadError, exceptions.InvalidLogFileMimeError, exceptions.LogFileIsEmptyError, ): diff --git a/reports/tests/test_services.py b/reports/tests/test_services.py index 34d25d4..466353a 100644 --- a/reports/tests/test_services.py +++ b/reports/tests/test_services.py @@ -47,6 +47,27 @@ def test_populate_log_report_tables_aggregates_log_files(self): } }, ) + LogFile.objects.create( + collection=collection, + path="/tmp/2026-05-10-error.log.gz", + stat_result={}, + hash="2" * 32, + status=choices.LOG_FILE_STATUS_INVALIDATED, + ) + LogFile.objects.create( + collection=collection, + path="/tmp/2026-05-10-corrupted.log.gz", + stat_result={}, + hash="3" * 32, + status=choices.LOG_FILE_STATUS_ERROR, + validation={ + "file_error": { + "code": "file_read_error", + "kind": "corrupted", + "stage": "catalog", + } + }, + ) result = log_report.populate_log_report_tables( year=2026, @@ -60,8 +81,10 @@ def test_populate_log_report_tables_aggregates_log_files(self): yearly = YearlyLogReport.objects.get(collection=collection) for report in [weekly, monthly, yearly]: - self.assertEqual(report.total_files, 1) + self.assertEqual(report.total_files, 3) self.assertEqual(report.validated_files, 1) + self.assertEqual(report.invalidated_files, 1) + self.assertEqual(report.errored_files, 1) self.assertEqual(report.lines_parsed, 10) self.assertEqual(report.valid_lines, 7) self.assertEqual(report.discarded_lines, 3) diff --git a/reports/wagtail_hooks.py b/reports/wagtail_hooks.py index 5e9e76b..e457889 100644 --- a/reports/wagtail_hooks.py +++ b/reports/wagtail_hooks.py @@ -19,6 +19,10 @@ def users_with_any_permission(self, actions): COMMON_LIST_DISPLAY = ( "total_files", + "created_files", + "validated_files", + "invalidated_files", + "errored_files", "pct_validated", "lines_parsed", "pct_valid_lines", diff --git a/requirements/base.txt b/requirements/base.txt index e28d18d..e6da798 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -63,7 +63,7 @@ minio==7.2.7 reverse-geocode==1.6 # https://pypi.org/project/reverse-geocode/ # SciELO Log Validator -git+https://github.com/scieloorg/scielo_log_validator@2.0.1#egg=scielo_log_validator +git+https://github.com/scieloorg/scielo_log_validator@2.0.2#egg=scielo_log_validator # SciELO Scholarly Data git+https://github.com/scieloorg/scielo_scholarly_data@v0.1.4#egg=scielo_scholarly_data