datastream: add basic OS logging

.github/workflows/tests.yml

@@ -29,4 +29,3 @@ jobs:
     uses: inveniosoftware/workflows/.github/workflows/tests-python.yml@master
     with:
       db-service: '[""]'
-      search-service: '[""]'

MANIFEST.in

@@ -23,7 +23,8 @@ recursive-include docs Makefile
 recursive-include examples *.py
 recursive-include examples *.sh
 recursive-include invenio_logging *.html
+recursive-include invenio_logging *.json
 recursive-include misc *.py
 recursive-include misc *.rst
 recursive-include tests *.py
-include .git-blame-ignore-revs
+include .git-blame-ignore-revs

invenio_logging/datastream.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Invenio OpenSearch Datastream Logging module."""
+
+from __future__ import absolute_import, print_function
+
+from importlib_metadata import entry_points
+
+from .datastreams.managers import LogManager
+from .ext import InvenioLoggingBase
+from .resources import LogsResource, LogsResourceConfig
+
+
+class InvenioLoggingDatastreams(InvenioLoggingBase):
+    """Invenio-Logging extension for OpenSearch Datastreams."""
+
+    def init_app(self, app):
+        """Initialize app.
+
+        :param app: An instance of :class:`~flask.Flask`.
+        """
+        self.init_manager()
+        self.init_resources(app)
+        self.load_builders()
+        app.extensions["invenio-logging-datastreams"] = self
+
+    def init_manager(self):
+        """Initialize the logging manager."""
+        manager = LogManager()
+        self.manager = manager
+
+    def load_builders(self):
+        """Load log builders from entry points."""
+        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/__init__.py

@@ -0,0 +1,9 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Invenio module for datastream logging management."""

invenio_logging/datastreams/app_logs/__init__.py

@@ -0,0 +1,16 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Loggigng module for app."""
+
+from .backends import SearchAppLogBackend
+from .builders import AppLogBuilder
+
+__all__ = (
+    "SearchAppLogBackend",
+    "AppLogBuilder",
+)

invenio_logging/datastreams/app_logs/backends.py

@@ -0,0 +1,18 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Loggigng module for app."""
+
+from invenio_logging.datastreams.backends import SearchBackend
+
+
+class SearchAppLogBackend(SearchBackend):
+    """Backend for storing app logs in datastreams."""
+
+    def __init__(self):
+        """Initialize backend for app logs."""
+        super().__init__(log_type="app")

invenio_logging/datastreams/app_logs/builders.py

@@ -0,0 +1,36 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Loggigng module for app."""
+
+from invenio_logging.datastreams.builders import LogBuilder
+
+from .backends import SearchAppLogBackend
+
+
+class AppLogBuilder(LogBuilder):
+    """Builder for structured app logs."""
+
+    type = "app"
+
+    backend_cls = SearchAppLogBackend
+
+    @classmethod
+    def build(cls, log_event):
+        """Build an app log event context."""
+        return cls.validate(log_event)
+
+    @classmethod
+    def send(cls, log_event):
+        """Send log event using the backend."""
+        cls.backend_cls().send(log_event)
+
+    @classmethod
+    def search(cls, query):
+        """Search logs."""
+        results = cls.backend_cls().search(query)
+        return cls.schema.load(results, many=True)

invenio_logging/datastreams/audit_logs/__init__.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Invenio module for audit-logs management."""
+
+from .backends import SearchAuditLogBackend
+from .builders import AuditLogBuilder
+
+__all__ = (
+    "SearchAuditLogBackend",
+    "AuditLogBuilder",
+)

invenio_logging/datastreams/audit_logs/backends.py

@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Invenio module for datastream logging management."""
+
+from invenio_logging.datastreams.backends import SearchBackend
+
+
+class SearchAuditLogBackend(SearchBackend):
+    """Backend for storing audit logs in datastreams."""
+
+    def __init__(self):
+        """Initialize backend for audit logs."""
+        super().__init__(log_type="audit")

invenio_logging/datastreams/audit_logs/builders.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Datastream Logging Builder module."""
+
+from invenio_logging.datastreams.builders import LogBuilder
+
+from .backends import SearchAuditLogBackend
+
+
+class AuditLogBuilder(LogBuilder):
+    """Builder for structured audit logs."""
+
+    type = "audit"
+
+    backend_cls = SearchAuditLogBackend
+
+    @classmethod
+    def build(cls, log_event):
+        """Build an audit log event context."""
+        return cls.validate(log_event)
+
+    @classmethod
+    def send(cls, log_event):
+        """Send log event using the backend."""
+        cls.backend_cls().send(log_event)
+
+    @classmethod
+    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

@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2025 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Invenio module for datastream backends."""
+
+
+from abc import ABC, abstractmethod
+
+from flask import current_app
+from invenio_search import current_search_client
+
+
+class LogBackend(ABC):
+    """Abstract base class for log backends."""
+
+    @abstractmethod
+    def send(self, log_type, log_event):
+        """Send a log event to the backend."""
+        raise NotImplementedError()
+
+
+class SearchBackend(LogBackend):
+    """Generic backend for storing logs in datastreams index."""
+
+    def __init__(self, log_type):
+        """
+        Initialize SearchBackend.
+
+        :param log_type: Type of log (e.g., "audit", "task", "system").
+        """
+        self.client = current_search_client
+        self.log_type = log_type
+        self.template_name = "datastream-log-v1.0.0"
+        self.index_name = f"logs-{log_type}"
+        self.search_fields = [
+            "message",
+            "event.action",
+            "user.id",
+            "user.email",
+            "resource.id",
+            "resource.parent.id",
+        ]
+
+        self._ensure_template_exists()
+
+    def _ensure_template_exists(self):
+        """Check if required template exists, enforce if missing."""
+        index_prefix = current_app.config.get("SEARCH_INDEX_PREFIX", "")
+        full_template_name = f"{index_prefix}{self.template_name}"
+        if not self.client.indices.exists_index_template(name=full_template_name):
+            raise RuntimeError(
+                f"Required template '{self.template_name}' is missing. "
+                "Ensure it is created before logging events."
+            )
+
+    def send(self, log_event):
+        """Send the log event to Search engine."""
+        try:
+            index_prefix = current_app.config.get("SEARCH_INDEX_PREFIX", "")
+            full_index_name = f"{index_prefix}{self.index_name}"
+            self.client.index(index=full_index_name, body=log_event)
+
+        except Exception as e:
+            current_app.logger.error(f"Failed to send log Search engine: {e}")
+            raise e
+
+    def search(self, query=None, size=10):
+        """
+        Search log events.
+
+        :param size: Number of results to return.
+        :return: List of log events that match the search query.
+        """
+        try:
+            index_prefix = current_app.config.get("SEARCH_INDEX_PREFIX", "")
+            full_index_name = f"{index_prefix}{self.index_name}"
+
+            search_query = {
+                "size": size,
+                "query": {
+                    "multi_match": {
+                        "query": query,
+                        "fields": self.search_fields,
+                        "operator": "and",
+                    }
+                },
+                "sort": [{"@timestamp": {"order": "desc"}}],
+            }
+            # TODO: add pagination?
+            response = self.client.search(index=full_index_name, body=search_query)
+            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
+
+    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