datastream: add basic OS logging

invenio_logging/datastream.py

@@ -12,9 +12,9 @@
 
 from importlib_metadata import entry_points
 
-from . import config
 from .datastreams.managers import LogManager
 from .ext import InvenioLoggingBase
+from .resources import LogsResource, LogsResourceConfig
 
 
 class InvenioLoggingDatastreams(InvenioLoggingBase):
@@ -25,11 +25,12 @@ def init_app(self, app):
 
         :param app: An instance of :class:`~flask.Flask`.
         """
-        self.init_manager(app)
+        self.init_manager()
+        self.init_resources(app)
         self.load_builders()
         app.extensions["invenio-logging-datastreams"] = self
 
-    def init_manager(self, app):
+    def init_manager(self):
         """Initialize the logging manager."""
         manager = LogManager()
         self.manager = manager
@@ -39,3 +40,10 @@ def load_builders(self):
         for ep in entry_points(group="invenio_logging.datastreams.builders"):
             builder_class = ep.load()
             self.manager.register_builder(ep.name, builder_class)
+
+    def init_resources(self, app):
+        """Init resources."""
+        self.resource = LogsResource(
+            config=LogsResourceConfig.build(app),
+            manager=self.manager,
+        )

invenio_logging/datastreams/audit_logs/builders.py

@@ -35,3 +35,9 @@ def search(cls, query):
         """Search logs."""
         results = cls.backend_cls().search(query)
         return cls.schema.dump(results, many=True)
+
+    @classmethod
+    def list(cls):
+        """List audit logs."""
+        results = cls.backend_cls().list()
+        return cls.schema.dump(results, many=True)

invenio_logging/datastreams/backends.py

@@ -89,7 +89,7 @@ def search(self, query=None, size=10):
                         "operator": "and",
                     }
                 },
-                "sort": [{"timestamp": {"order": "desc"}}],
+                "sort": [{"@timestamp": {"order": "desc"}}],
             }
             # TODO: add pagination?
             response = self.client.search(index=full_index_name, body=search_query)
@@ -98,3 +98,24 @@ def search(self, query=None, size=10):
         except Exception as e:
             current_app.logger.error(f"Failed to search logs: {e}")
             raise e
+
+    def list(self, size=10):
+        """
+        List log events.
+
+        :param size: Number of results to return.
+        :return: List of log events.
+        """
+        try:
+            index_prefix = current_app.config.get("SEARCH_INDEX_PREFIX", "")
+            full_index_name = f"{index_prefix}{self.index_name}"
+
+            response = self.client.search(
+                index=full_index_name,
+                body={"size": size, "sort": [{"@timestamp": {"order": "desc"}}]},
+            )
+            return [hit["_source"] for hit in response.get("hits", {}).get("hits", [])]
+
+        except Exception as e:
+            current_app.logger.error(f"Failed to search logs: {e}")
+            raise e

invenio_logging/datastreams/managers.py

@@ -43,7 +43,9 @@ def log(self, log_event, async_mode=True):
             log_builder.send(log)
 
     def search(self, log_type, query):
-        """Search logs."""
+        """
+        Search for logs using the correct builder.
+        """
         if log_type not in self.builders:
             raise ValueError(
                 f"No log builder found for type '{log_type}'. Available types: {self.builders.keys()}"
@@ -52,6 +54,18 @@ def search(self, log_type, query):
         log_builder = self.builders[log_type]
         return log_builder.search(query)
 
+    def list(self, log_type):
+        """
+        List logs using the correct builder.
+        """
+        if log_type not in self.builders:
+            raise ValueError(
+                f"No log builder found for type '{log_type}'. Available types: {self.builders.keys()}"
+            )
+
+        log_builder = self.builders[log_type]
+        return log_builder.list()
+
     def register_builder(self, log_type, builder_class):
         """Register a log builder."""
         self.builders[log_type] = builder_class

invenio_logging/errors.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Requests is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Custom exceptions used in the Invenio-Logging module."""
+
+
+class InvalidLogQueryError(Exception):
+    """Error raised when an invalid query is made on logging resources."""
+
+    def __init__(self, message="Invalid log query parameters provided."):
+        """Initialize error with a default message."""
+        self.message = message
+        super().__init__(self.message)

invenio_logging/resources/__init__.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Requests is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Resources module."""
+
+from .resource import LogsResource
+from .config import LogsResourceConfig
+
+__all__ = (
+    "LogsResource",
+    "LogsResourceConfig",
+)

invenio_logging/resources/config.py

@@ -0,0 +1,56 @@
+from marshmallow import fields
+
+from invenio_records_resources.resources import SearchRequestArgsSchema, RecordResourceConfig
+from invenio_records_resources.services.base.config import ConfiguratorMixin
+
+from flask_resources import HTTPJSONException, create_error_handler
+from ..errors import InvalidLogQueryError
+
+#
+# Request args
+#
+class LogSearchRequestArgsSchema(SearchRequestArgsSchema):
+    """Search parameters for logs."""
+
+    resource_id = fields.String()
+    resource_type = fields.String()
+    user_id = fields.String()
+    action = fields.String()
+
+error_handlers = {
+    InvalidLogQueryError: create_error_handler(
+        lambda e: HTTPJSONException(code=400, description=str(e))
+    ),
+}
+
+#
+# Resource config
+#
+class LogsResourceConfig(RecordResourceConfig, ConfiguratorMixin):
+    """Logs resource configuration."""
+
+    blueprint_name = "logs"
+    url_prefix = "/logs"
+
+    routes = {
+        "list": "/",
+        "item": "/<id>",
+    }
+
+    request_view_args = {
+        "resource_id": fields.String(),
+        # "resource_type": fields.String(), # TODO: Add direct querying via other search parameters?
+        # "user_id": fields.String(),
+        # "action": fields.String(),
+    }
+
+    request_search_args = LogSearchRequestArgsSchema
+
+    error_handlers = error_handlers
+
+    response_handlers = {
+        "application/vnd.inveniordm.v1+json": RecordResourceConfig.response_handlers[
+            "application/json"
+        ],
+        **RecordResourceConfig.response_handlers,
+    }

invenio_logging/resources/resource.py

@@ -0,0 +1,77 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio-Logging is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Logs resource."""
+
+from flask import g
+from flask_resources import resource_requestctx, response_handler, route
+from flask_resources import Resource
+from invenio_records_resources.resources.records.resource import (
+    # request_data,
+    request_extra_args,
+    # request_headers,
+    request_search_args,
+    request_view_args,
+)
+from invenio_records_resources.resources.records.utils import search_preference
+
+
+#
+# Resource
+#
+class LogsResource(Resource):
+    """Resource for logs (audit and job logs)."""
+
+    def __init__(self, config, manager):
+        """Constructor."""
+        super(LogsResource, self).__init__(config)
+        self.manager = manager
+
+    def create_blueprint(self, **options):
+        """Create the blueprint."""
+        options["url_prefix"] = ""
+        return super().create_blueprint(**options)
+
+    def create_url_rules(self):
+        """Create the URL rules for the log resource."""
+        routes = self.config.routes
+
+        def p(route):
+            """Prefix a route with the URL prefix."""
+            return f"{self.config.url_prefix}{route}"
+
+        return [
+            route("GET", p(routes["list"]), self.search),
+            route("GET", p(routes["item"]), self.read),
+        ]
+
+    @request_extra_args
+    @request_search_args
+    @request_view_args
+    @response_handler(many=True)
+    def search(self):
+        """Perform a search over the logs."""
+        hits = self.manager.search(
+            identity=g.identity,
+            params=resource_requestctx.args,
+            search_preference=search_preference(),
+            expand=resource_requestctx.args.get("expand", False),
+        )
+        return hits.to_dict(), 200
+
+    @request_extra_args
+    @request_view_args
+    @response_handler()
+    def read(self):
+        """Read a specific log entry."""
+        item = self.manager.read_all(
+            id_=resource_requestctx.view_args["id"],
+            identity=g.identity,
+            expand=resource_requestctx.args.get("expand", False),
+        )
+        return item.to_dict(), 200