Registra e recupera logs ilegíveis na versão 2.3.6

VERSION

@@ -1 +1 @@
-2.3.5
+2.3.6

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"

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)

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,30 +53,154 @@ 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)
             if extension.lower() not in supported_extensions:
                 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()

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: