feat: Logging is working, and we can search through the data

Author: Robin-Van-de-Merghel

diracx-core/src/diracx/core/models.py

@@ -324,11 +324,12 @@ class JobCommand(BaseModel):
 
 
 class LogLine(BaseModel):
-    line_no: int
-    line: str
+    timestamp: str
+    severity: str
+    message: str
+    scope: str
 
 
 class LogMessage(BaseModel):
     pilot_stamp: str
     lines: list[LogLine]
-    vo: str

diracx-db/src/diracx/db/os/pilot_logs.py

@@ -7,24 +7,15 @@ class PilotLogsDB(BaseOSDB):
     fields = {
         "PilotStamp": {"type": "keyword"},
         "PilotID": {"type": "long"},
-        "SubmissionTime": {"type": "date"},
-        "LineNumber": {"type": "long"},
+        "Severity": {"type": "keyword"},
         "Message": {"type": "text"},
         "VO": {"type": "keyword"},
-        "timestamp": {"type": "date"},
+        "TimeStamp": {"type": "date_nanos"},
+        "Scope": {"type": "keyword"},
     }
     index_prefix = "pilot_logs"
 
     def index_name(self, vo: str, doc_id: int) -> str:
         # TODO decide how to define the index name
         # use pilot ID
-        return f"{self.index_prefix}_{doc_id // 1e6:.0f}"
-
-
-async def search_message(db: PilotLogsDB, search_params: list[dict]):
-
-    return await db.search(
-        ["Message"],
-        search_params,
-        [{"parameter": "LineNumber", "direction": "asc"}],
-    )
+        return f"{self.index_prefix}_{doc_id}"

diracx-db/src/diracx/db/os/utils.py

@@ -200,13 +200,22 @@ async def upsert(self, vo: str, doc_id: int, document: Any) -> None:
 
     async def bulk_insert(self, index_name: str, docs: list[dict[str, Any]]) -> None:
         """Bulk inserting to database."""
-        n_inserted = await async_bulk(
+        n_inserted, failed = await async_bulk(
             self.client, actions=[doc | {"_index": index_name} for doc in docs]
         )
-        logger.info("Inserted %d documents to %r", n_inserted, index_name)
+        logger.info("Inserted %d documents to %s", n_inserted, index_name)
+
+        if failed:
+            logger.error("Fail to insert %d documents to %s", failed, index_name)
 
     async def search(
-        self, parameters, search, sorts, *, per_page: int = 100, page: int | None = None
+        self,
+        parameters,
+        search,
+        sorts,
+        *,
+        per_page: int = 10000,
+        page: int | None = None,
     ) -> list[dict[str, Any]]:
         """Search the database for matching results.
 
@@ -221,7 +230,12 @@ async def search(
         for sort in sorts:
             field_name = sort["parameter"]
             field_type = self.fields.get(field_name, {}).get("type")
-            require_type("sort", field_name, field_type, {"keyword", "long", "date"})
+            require_type(
+                "sort",
+                field_name,
+                field_type,
+                {"keyword", "long", "date", "date_nanos"},
+            )
             body["sort"].append({field_name: {"order": sort["direction"]}})
 
         params = {}
@@ -239,7 +253,7 @@ async def search(
             for field_name in hit:
                 if field_name not in self.fields:
                     continue
-                if self.fields[field_name]["type"] == "date":
+                if self.fields[field_name]["type"] in ["date", "date_nanos"]:
                     hit[field_name] = datetime.strptime(
                         hit[field_name], "%Y-%m-%dT%H:%M:%S.%f%z"
                     )
@@ -286,30 +300,46 @@ def apply_search_filters(db_fields, search):
         match operator := query["operator"]:
             case "eq":
                 require_type(
-                    operator, field_name, field_type, {"keyword", "long", "date"}
+                    operator,
+                    field_name,
+                    field_type,
+                    {"keyword", "long", "date", "date_nanos"},
                 )
                 result["must"].append({"term": {field_name: {"value": query["value"]}}})
             case "neq":
                 require_type(
-                    operator, field_name, field_type, {"keyword", "long", "date"}
+                    operator,
+                    field_name,
+                    field_type,
+                    {"keyword", "long", "date", "date_nanos"},
                 )
                 result["must_not"].append(
                     {"term": {field_name: {"value": query["value"]}}}
                 )
             case "gt":
-                require_type(operator, field_name, field_type, {"long", "date"})
+                require_type(
+                    operator, field_name, field_type, {"long", "date", "date_nanos"}
+                )
                 result["must"].append({"range": {field_name: {"gt": query["value"]}}})
             case "lt":
-                require_type(operator, field_name, field_type, {"long", "date"})
+                require_type(
+                    operator, field_name, field_type, {"long", "date", "date_nanos"}
+                )
                 result["must"].append({"range": {field_name: {"lt": query["value"]}}})
             case "in":
                 require_type(
-                    operator, field_name, field_type, {"keyword", "long", "date"}
+                    operator,
+                    field_name,
+                    field_type,
+                    {"keyword", "long", "date", "date_nanos"},
                 )
                 result["must"].append({"terms": {field_name: query["values"]}})
             case "not in":
                 require_type(
-                    operator, field_name, field_type, {"keyword", "long", "date"}
+                    operator,
+                    field_name,
+                    field_type,
+                    {"keyword", "long", "date", "date_nanos"},
                 )
                 result["must_not"].append({"terms": {field_name: query["values"]}})
             # TODO: Implement like and ilike

diracx-logic/src/diracx/logic/pilots/logging.py

@@ -2,68 +2,84 @@
 
 import logging
 
-from diracx.core.models import LogMessage, ScalarSearchOperator, ScalarSearchSpec
-from diracx.db.os.pilot_logs import PilotLogsDB, search_message
+from diracx.core.exceptions import PilotNotFoundError
+from diracx.core.models import (
+    LogMessage,
+    SearchParams,
+    SortDirection,
+)
+from diracx.db.os.pilot_logs import PilotLogsDB
 from diracx.db.sql.pilot_agents.db import PilotAgentsDB
 
 logger = logging.getLogger(__name__)
 
 
+MAX_PER_PAGE = 10000
+
+
 async def send_message(
     data: LogMessage,
     pilot_logs_db: PilotLogsDB,
     pilot_agents_db: PilotAgentsDB,
-) -> int:
-
-    # get the pilot ID corresponding to a given pilot stamp, expecting exactly one row:
-    search_params = ScalarSearchSpec(
-        parameter="PilotStamp",
-        operator=ScalarSearchOperator.EQUAL,
-        value=data.pilot_stamp,
-    )
-
-    total, result = await pilot_agents_db.search(
-        ["PilotID", "VO", "SubmissionTime"], [search_params], []
-    )
-    if total != 1:
-        logger.error(
-            "Cannot determine PilotID for requested PilotStamp: %r, (%d candidates)",
-            data.pilot_stamp,
-            total,
-        )
-        raise Exception(f"Number of rows !=1 {total}")
+    vo: str,
+):
+    try:
+        pilot_ids = await pilot_agents_db.get_pilot_ids_by_stamps([data.pilot_stamp])
+        pilot_id = pilot_ids[0]  # Semantic
+    except PilotNotFoundError:
+        # If a pilot is not found, then we still store the data (to not lost it)
+        # We log it as it's not supposed to happen
+        pilot_id = -1  # To detect
 
-    pilot_id, vo, submission_time = (
-        result[0]["PilotID"],
-        result[0]["VO"],
-        result[0]["SubmissionTime"],
-    )
     docs = []
     for line in data.lines:
         docs.append(
             {
                 "PilotStamp": data.pilot_stamp,
                 "PilotID": pilot_id,
-                "SubmissionTime": submission_time,
                 "VO": vo,
-                "LineNumber": line.line_no,
-                "Message": line.line,
+                "Severity": line.severity,
+                "Message": line.message,
+                "TimeStamp": line.timestamp,
+                "Scope": line.scope,
             }
         )
     # bulk insert pilot logs to OpenSearch DB:
     await pilot_logs_db.bulk_insert(pilot_logs_db.index_name(vo, pilot_id), docs)
-    return pilot_id
 
 
-async def get_logs(
-    pilot_id: int,
-    db: PilotLogsDB,
+async def search_logs(
+    vo: str,
+    body: SearchParams | None,
+    per_page: int,
+    page: int,
+    pilot_logs_db: PilotLogsDB,
 ) -> list[dict]:
+    """Retrieve logs from OpenSearch for a given PilotStamp."""
+    # Apply a limit to per_page to prevent abuse of the API
+    if per_page > MAX_PER_PAGE:
+        per_page = MAX_PER_PAGE
 
-    search_params = [{"parameter": "PilotID", "operator": "eq", "value": pilot_id}]
+    if body is None:
+        body = SearchParams()
 
-    result = await search_message(db, search_params)
+    search = body.search
+    parameters = body.parameters
+    sorts = body.sort
 
-    if not result:
-        return [{"Message": f"No logs for pilot ID = {pilot_id}"}]
-    return result
+    # Add the vo to make sure that we filter for pilots we can see
+    # TODO: Test it
+    search = search + [
+        {
+            "parameter": "VO",
+            "operator": "eq",
+            "value": vo,
+        }
+    ]
+
+    if not sorts:
+        sorts = [{"parameter": "TimeStamp", "direction": SortDirection("asc")}]
+
+    return await pilot_logs_db.search(
+        parameters=parameters, search=search, sorts=sorts, per_page=per_page, page=page
+    )

diracx-routers/src/diracx/routers/pilots/access_policies.py

@@ -8,8 +8,12 @@
 from fastapi import Depends, HTTPException, status
 
 from diracx.core.exceptions import PilotNotFoundError
-from diracx.core.models import ScalarSearchOperator, ScalarSearchSpec
-from diracx.core.properties import NORMAL_USER, TRUSTED_HOST
+from diracx.core.properties import (
+    GENERIC_PILOT,
+    LIMITED_DELEGATION,
+    NORMAL_USER,
+    TRUSTED_HOST,
+)
 from diracx.db.sql import PilotAgentsDB
 from diracx.routers.access_policies import BaseAccessPolicy
 from diracx.routers.utils.users import AuthorizedUserInfo
@@ -25,9 +29,9 @@ class ActionType(StrEnum):
     # Read some pilot info
     READ_PILOT_FIELDS = auto()
     #: Create/update pilot log records
-    CREATE = auto()
+    CREATE_LOG = auto()
     #: Search
-    QUERY = auto()
+    QUERY_LOGS = auto()
 
 
 class PilotManagementAccessPolicy(BaseAccessPolicy):
@@ -49,7 +53,6 @@ async def policy(
         action: ActionType | None = None,
     ):
         assert action, "action is a mandatory parameter"
-
         if action == ActionType.READ_PILOT_FIELDS:
             if NORMAL_USER in user_info.properties:
                 return
@@ -132,60 +135,47 @@ async def policy(policy_name: str, user_info: AuthorizedUserInfo):
 
 class PilotLogsAccessPolicy(BaseAccessPolicy):
     """Rules:
-    Only NORMAL_USER in a correct VO and a diracAdmin VO member can query log records.
+    Only NORMAL_USER in a correct VO can query log records
     All other actions and users are explicitly denied access.
     """
 
+    # TODO: Better doc
+
     @staticmethod
     async def policy(
         policy_name: str,
         user_info: AuthorizedUserInfo,
         /,
         *,
         action: ActionType | None = None,
-        pilot_agents_db: PilotAgentsDB | None = None,
-        pilot_id: int | None = None,
+        pilot_db: PilotAgentsDB | None = None,
     ):
-        assert pilot_agents_db
-        if action is None:
-            raise HTTPException(
-                status.HTTP_400_BAD_REQUEST, detail="Action is a mandatory argument"
-            )
-        elif action == ActionType.QUERY:
-            if pilot_id is None:
-                logger.error("Pilot ID value is not provided (None)")
-                raise HTTPException(
-                    status.HTTP_400_BAD_REQUEST,
-                    detail=f"PilotID not provided: {pilot_id}",
-                )
-            search_params = ScalarSearchSpec(
-                parameter="PilotID",
-                operator=ScalarSearchOperator.EQUAL,
-                value=pilot_id,
-            )
 
-            total, result = await pilot_agents_db.search(["VO"], [search_params], [])
-            # we expect exactly one row.
-            if total != 1:
-                logger.error(
-                    "Cannot determine VO for requested PilotID: %d, found %d candidates.",
-                    pilot_id,
-                    total,
-                )
-                raise HTTPException(
-                    status.HTTP_400_BAD_REQUEST, detail=f"PilotID not found: {pilot_id}"
-                )
-            vo = result[0]["VO"]
+        assert pilot_db
+        if action == ActionType.CREATE_LOG:
+            pilot_info = user_info  # semantic
 
-            if user_info.vo == "diracAdmin":
+            if GENERIC_PILOT in pilot_info.properties:
                 return
 
-            if NORMAL_USER in user_info.properties and user_info.vo == vo:
+            if LIMITED_DELEGATION in pilot_info.properties:
+                return
+
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You must be a pilot to create logs.",
+            )
+
+        elif action == ActionType.QUERY_LOGS:
+            # TODO: See if we move this elsewhere to separate from pilots
+            # TODO: To see if we can access a VO, we add a new rule a the end in the logic
+            # -> Do we add it here? (another verification)
+            if NORMAL_USER in user_info.properties:
                 return
 
             raise HTTPException(
                 status.HTTP_403_FORBIDDEN,
-                detail="You don't have permission to access this pilot's log.",
+                detail="You don't have permission to access pilots log",
             )
         else:
             raise NotImplementedError(action)