Move crawl and QA logs to new mongo collection (#2791)

Fixes #2765 

(This is a necessary part of ensuring mongo documents don't exceed 16MB.
Hopefully it's also sufficient but we'll need to see in practice if
there are other fields that need to be separated out.)

This PR moves crawl and QA run logs into a separate `crawl_logs` mongo
collection.

It adds a new backend module (without a distinct API router, but the
crawls module is getting quite large and it seemed to make sense to add
a separate module for the new mongo collection), as well as a migration
to move crawl logs from `Crawl` objects into the new collection. The
existing nightly test for crawl error logs is fleshed out, and a new
nightly test added for behavior logs.

---------
Co-authored-by: Ilya Kreymer <[email protected]>
Co-authored-by: Ilya Kreymer <[email protected]>

Author: tw4l

backend/btrixcloud/crawl_logs.py

@@ -0,0 +1,181 @@
+"""crawl logs"""
+
+from typing import TYPE_CHECKING, Any, Optional, Dict, Tuple, List
+
+import json
+from uuid import UUID, uuid4
+
+from fastapi import HTTPException
+import pymongo
+
+from .models import CrawlLogLine, Organization
+from .pagination import DEFAULT_PAGE_SIZE
+
+if TYPE_CHECKING:
+    from .orgs import OrgOps
+else:
+    OrgOps = object
+
+
+# ============================================================================
+class CrawlLogOps:
+    """crawl log management"""
+
+    org_ops: OrgOps
+
+    # pylint: disable=too-many-locals, too-many-arguments, invalid-name
+
+    def __init__(self, mdb, org_ops):
+        self.logs = mdb["crawl_logs"]
+        self.org_ops = org_ops
+
+    async def init_index(self):
+        """init index for crawl logs"""
+        await self.logs.create_index(
+            [
+                ("crawlId", pymongo.HASHED),
+                ("oid", pymongo.ASCENDING),
+                ("qaRunId", pymongo.ASCENDING),
+                ("timestamp", pymongo.ASCENDING),
+            ]
+        )
+        await self.logs.create_index(
+            [
+                ("crawlId", pymongo.HASHED),
+                ("oid", pymongo.ASCENDING),
+                ("qaRunId", pymongo.ASCENDING),
+                ("logLevel", pymongo.ASCENDING),
+            ]
+        )
+        await self.logs.create_index(
+            [
+                ("crawlId", pymongo.HASHED),
+                ("oid", pymongo.ASCENDING),
+                ("qaRunId", pymongo.ASCENDING),
+                ("context", pymongo.ASCENDING),
+            ]
+        )
+        await self.logs.create_index(
+            [
+                ("crawlId", pymongo.HASHED),
+                ("oid", pymongo.ASCENDING),
+                ("qaRunId", pymongo.ASCENDING),
+                ("message", pymongo.ASCENDING),
+            ]
+        )
+
+    async def add_log_line(
+        self,
+        crawl_id: str,
+        oid: UUID,
+        log_line: str,
+        qa_run_id: Optional[str] = None,
+    ) -> bool:
+        """add crawl log line to database"""
+        try:
+            log_dict = json.loads(log_line)
+
+            # Ensure details are a dictionary
+            # If they are a list, convert to a dict
+            details = None
+            log_dict_details = log_dict.get("details")
+            if log_dict_details:
+                if isinstance(log_dict_details, dict):
+                    details = log_dict_details
+                else:
+                    details = {"items": log_dict_details}
+
+            log_to_add = CrawlLogLine(
+                id=uuid4(),
+                crawlId=crawl_id,
+                oid=oid,
+                qaRunId=qa_run_id,
+                timestamp=log_dict["timestamp"],
+                logLevel=log_dict["logLevel"],
+                context=log_dict["context"],
+                message=log_dict["message"],
+                details=details,
+            )
+            res = await self.logs.insert_one(log_to_add.to_dict())
+            return res is not None
+        # pylint: disable=broad-exception-caught
+        except Exception as err:
+            print(
+                f"Error adding log line for crawl {crawl_id} to database: {err}",
+                flush=True,
+            )
+            return False
+
+    async def get_crawl_logs(
+        self,
+        org: Organization,
+        crawl_id: str,
+        page_size: int = DEFAULT_PAGE_SIZE,
+        page: int = 1,
+        sort_by: str = "timestamp",
+        sort_direction: int = -1,
+        contexts: Optional[List[str]] = None,
+        log_levels: Optional[List[str]] = None,
+        qa_run_id: Optional[str] = None,
+    ) -> Tuple[list[CrawlLogLine], int]:
+        """list all logs for particular crawl"""
+        # pylint: disable=too-many-locals, duplicate-code
+
+        # Zero-index page for query
+        page = page - 1
+        skip = page_size * page
+
+        match_query: Dict[str, Any] = {
+            "oid": org.id,
+            "crawlId": crawl_id,
+            "qaRunId": qa_run_id,
+        }
+
+        if contexts:
+            match_query["context"] = {"$in": contexts}
+
+        if log_levels:
+            match_query["logLevel"] = {"$in": log_levels}
+
+        aggregate: List[Dict[str, Any]] = [{"$match": match_query}]
+
+        if sort_by:
+            if sort_by not in (
+                "timestamp",
+                "logLevel",
+                "context",
+                "message",
+            ):
+                raise HTTPException(status_code=400, detail="invalid_sort_by")
+            if sort_direction not in (1, -1):
+                raise HTTPException(status_code=400, detail="invalid_sort_direction")
+
+            aggregate.extend([{"$sort": {sort_by: sort_direction}}])
+
+        aggregate.extend(
+            [
+                {
+                    "$facet": {
+                        "items": [
+                            {"$skip": skip},
+                            {"$limit": page_size},
+                        ],
+                        "total": [{"$count": "count"}],
+                    }
+                },
+            ]
+        )
+
+        cursor = self.logs.aggregate(aggregate)
+        results = await cursor.to_list(length=1)
+        result = results[0]
+        items = result["items"]
+
+        try:
+            total = int(result["total"][0]["count"])
+        except (IndexError, ValueError):
+            total = 0
+
+        log_lines = [CrawlLogLine.from_dict(res) for res in items]
+
+        return log_lines, total

backend/btrixcloud/crawls.py

@@ -10,7 +10,7 @@
 from datetime import datetime
 from uuid import UUID
 
-from typing import Optional, List, Dict, Union, Any, Sequence, AsyncIterator
+from typing import Optional, List, Dict, Union, Any, Sequence, AsyncIterator, Tuple
 
 from fastapi import Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
@@ -22,14 +22,14 @@
 from .utils import (
     dt_now,
     date_to_str,
-    parse_jsonl_log_messages,
     stream_dict_list_as_csv,
     validate_regexes,
     scale_from_browser_windows,
     browser_windows_from_scale,
 )
 from .basecrawls import BaseCrawlOps
 from .crawlmanager import CrawlManager
+from .crawl_logs import CrawlLogOps
 from .models import (
     UpdateCrawl,
     DeleteCrawlList,
@@ -66,6 +66,7 @@
     CrawlScaleResponse,
     CrawlQueueResponse,
     MatchCrawlQueueResponse,
+    CrawlLogLine,
 )
 
 
@@ -80,9 +81,10 @@ class CrawlOps(BaseCrawlOps):
 
     crawl_manager: CrawlManager
 
-    def __init__(self, crawl_manager: CrawlManager, *args):
+    def __init__(self, crawl_manager: CrawlManager, log_ops: CrawlLogOps, *args):
         super().__init__(*args)
         self.crawl_manager = crawl_manager
+        self.log_ops = log_ops
         self.crawl_configs.set_crawl_ops(self)
         self.colls.set_crawl_ops(self)
         self.event_webhook_ops.set_crawl_ops(self)
@@ -669,31 +671,6 @@ async def is_upload(self, crawl_id: str):
             return False
         return res.get("type") == "upload"
 
-    async def add_crawl_error(
-        self,
-        crawl_id: str,
-        is_qa: bool,
-        error: str,
-    ) -> bool:
-        """add crawl error from redis to mongodb errors field"""
-        prefix = "" if not is_qa else "qa."
-
-        res = await self.crawls.find_one_and_update(
-            {"_id": crawl_id}, {"$push": {f"{prefix}errors": error}}
-        )
-        return res is not None
-
-    async def add_crawl_behavior_log(
-        self,
-        crawl_id: str,
-        log_line: str,
-    ) -> bool:
-        """add crawl behavior log from redis to mongodb behaviorLogs field"""
-        res = await self.crawls.find_one_and_update(
-            {"_id": crawl_id}, {"$push": {"behaviorLogs": log_line}}
-        )
-        return res is not None
-
     async def add_crawl_file(
         self, crawl_id: str, is_qa: bool, crawl_file: CrawlFile, size: int
     ) -> bool:
@@ -1141,6 +1118,31 @@ async def get_qa_run_aggregate_stats(
             textMatch=text_results,
         )
 
+    async def get_crawl_logs(
+        self,
+        org: Organization,
+        crawl_id: str,
+        page_size: int = DEFAULT_PAGE_SIZE,
+        page: int = 1,
+        sort_by: str = "timestamp",
+        sort_direction: int = 1,
+        contexts: Optional[List[str]] = None,
+        log_levels: Optional[List[str]] = None,
+        qa_run_id: Optional[str] = None,
+    ) -> Tuple[list[CrawlLogLine], int]:
+        """get crawl logs"""
+        return await self.log_ops.get_crawl_logs(
+            org,
+            crawl_id,
+            page_size=page_size,
+            page=page,
+            sort_by=sort_by,
+            sort_direction=sort_direction,
+            contexts=contexts,
+            log_levels=log_levels,
+            qa_run_id=qa_run_id,
+        )
+
 
 # ============================================================================
 async def recompute_crawl_file_count_and_size(crawls, crawl_id: str):
@@ -1162,11 +1164,13 @@ async def recompute_crawl_file_count_and_size(crawls, crawl_id: str):
 
 # ============================================================================
 # pylint: disable=too-many-arguments, too-many-locals, too-many-statements
-def init_crawls_api(crawl_manager: CrawlManager, app, user_dep, *args):
+def init_crawls_api(
+    crawl_manager: CrawlManager, crawl_log_ops: CrawlLogOps, app, user_dep, *args
+):
     """API for crawl management, including crawl done callback"""
     # pylint: disable=invalid-name, duplicate-code
 
-    ops = CrawlOps(crawl_manager, *args)
+    ops = CrawlOps(crawl_manager, crawl_log_ops, *args)
 
     org_viewer_dep = ops.orgs.org_viewer_dep
     org_crawl_dep = ops.orgs.org_crawl_dep
@@ -1694,15 +1698,20 @@ async def get_crawl_errors(
         pageSize: int = DEFAULT_PAGE_SIZE,
         page: int = 1,
         org: Organization = Depends(org_viewer_dep),
+        sortBy: str = "timestamp",
+        sortDirection: int = 1,
     ):
-        crawl = await ops.get_crawl(crawl_id, org)
-
-        skip = (page - 1) * pageSize
-        upper_bound = skip + pageSize
-
-        errors = crawl.errors[skip:upper_bound] if crawl.errors else []
-        parsed_errors = parse_jsonl_log_messages(errors)
-        return paginated_format(parsed_errors, len(crawl.errors or []), page, pageSize)
+        log_lines, total = await ops.get_crawl_logs(
+            org,
+            crawl_id,
+            page_size=pageSize,
+            page=page,
+            sort_by=sortBy,
+            sort_direction=sortDirection,
+            log_levels=["error", "fatal"],
+            qa_run_id=None,
+        )
+        return paginated_format(log_lines, total, page, pageSize)
 
     @app.get(
         "/orgs/{oid}/crawls/{crawl_id}/behaviorLogs",
@@ -1714,18 +1723,19 @@ async def get_crawl_behavior_logs(
         pageSize: int = DEFAULT_PAGE_SIZE,
         page: int = 1,
         org: Organization = Depends(org_viewer_dep),
+        sortBy: str = "timestamp",
+        sortDirection: int = 1,
     ):
-        crawl = await ops.get_crawl(crawl_id, org)
-
-        skip = (page - 1) * pageSize
-        upper_bound = skip + pageSize
-
-        behavior_logs = (
-            crawl.behaviorLogs[skip:upper_bound] if crawl.behaviorLogs else []
-        )
-        parsed_logs = parse_jsonl_log_messages(behavior_logs)
-        return paginated_format(
-            parsed_logs, len(crawl.behaviorLogs or []), page, pageSize
+        log_lines, total = await ops.get_crawl_logs(
+            org,
+            crawl_id,
+            page_size=pageSize,
+            page=page,
+            sort_by=sortBy,
+            sort_direction=sortDirection,
+            contexts=["behavior", "behaviorScript", "behaviorScriptCustom"],
+            qa_run_id=None,
         )
+        return paginated_format(log_lines, total, page, pageSize)
 
     return ops