add and register ng connectors (#850)

Author: ekneg54

logprep/ng/abc/output.py

@@ -0,0 +1,131 @@
+"""This module provides the abstract base class for all output endpoints.
+New output endpoint types are created by implementing it.
+"""
+
+import threading
+from abc import abstractmethod
+from copy import deepcopy
+from typing import Any, Callable
+
+from attrs import define, field, validators
+
+from logprep.abc.connector import Connector
+from logprep.abc.exceptions import LogprepException
+from logprep.ng.abc.event import Event
+
+
+class OutputError(LogprepException):
+    """Base class for Output related exceptions."""
+
+    def __init__(self, output: "Output", message: str) -> None:
+        output.metrics.number_of_errors += 1
+        super().__init__(f"{self.__class__.__name__} in {output.describe()}: {message}")
+
+
+class OutputWarning(LogprepException):
+    """Base class for Output related warnings."""
+
+    def __init__(self, output: "Output", message: str) -> None:
+        output.metrics.number_of_warnings += 1
+        super().__init__(f"{self.__class__.__name__} in {output.describe()}: {message}")
+
+
+class CriticalOutputError(OutputError):
+    """A significant error occurred - log and don't process the event."""
+
+    __match_args__ = ("raw_input",)
+
+    def __init__(self, output: "Output", message: str, raw_input: Any) -> None:
+        super().__init__(output, f"{message} -> event was written to error output if configured")
+        self.raw_input = deepcopy(raw_input)
+        self.message = message
+
+
+class FatalOutputError(OutputError):
+    """Must not be caught."""
+
+
+class Output(Connector):
+    """Connect to a output destination."""
+
+    @define(kw_only=True)
+    class Metrics(Connector.Metrics):
+        """Tracks statistics about this connector"""
+
+    @define(kw_only=True)
+    class Config(Connector.Config):
+        """output config parameters"""
+
+        default: bool = field(validator=validators.instance_of(bool), default=True)
+        """ (Optional) if :code:`false` the event are not delivered to this output.
+        But this output can be called as output for extra_data.
+        """
+
+    @property
+    def default(self):
+        """returns the default parameter"""
+        return self._config.default
+
+    @property
+    def metric_labels(self) -> dict:
+        """Return the metric labels for this component."""
+        return {
+            "component": "output",
+            "description": self.describe(),
+            "type": self._config.type,
+            "name": self.name,
+        }
+
+    def __init__(self, name: str, configuration: "Connector.Config"):
+        super().__init__(name, configuration)
+        self.input_connector = None
+        self.lock = threading.Lock()
+
+    @abstractmethod
+    def store(self, event: Event) -> None:
+        """Store the event in the output destination.
+
+        Parameters
+        ----------
+        event : dict
+           Processed log event that will be stored.
+        """
+
+    @abstractmethod
+    def store_custom(self, event: Event, target: str) -> None:
+        """Store the event in the output destination.
+
+        Parameters
+        ----------
+        event : dict
+           Processed log event that will be stored.
+        target : str
+            Custom target for the event.
+        """
+
+    @abstractmethod
+    def flush(self):
+        """Write the backlog to the output destination.
+        Needs to be implemented in child classes to ensure
+        that the backlog is written to the output destination.
+        """
+
+    @staticmethod
+    def _handle_errors(func: Callable) -> Callable:
+        """Decorator to handle errors during the store process."""
+
+        def wrapper(self, *args, **kwargs):
+            event = args[0] if args else kwargs.get("event")
+            try:
+                func(self, *args, **kwargs)
+            except Exception as e:  # pylint: disable=broad-except
+                event.errors.append(e)
+                self.metrics.number_of_errors += 1
+                event.state.next_state(success=False)
+
+        return wrapper
+
+    def shut_down(self) -> None:
+        """Shut down the output connector."""
+        self.flush()
+        super().shut_down()

logprep/ng/connector/console/__init__.py

@@ -0,0 +1,47 @@
+"""
+ConsoleOutput
+=============
+
+This section describes the ConsoleOutput, which pretty prints documents to the
+console and can be used for testing.
+
+Example
+^^^^^^^
+..  code-block:: yaml
+    :linenos:
+
+    output:
+      my_console_output:
+        type: console_output
+"""
+
+import sys
+from pprint import pprint
+
+from logprep.ng.abc.event import Event
+from logprep.ng.abc.output import Output
+
+
+class ConsoleOutput(Output):
+    """A console output that pretty prints documents instead of storing them."""
+
+    @Output._handle_errors
+    def store(self, event: Event) -> None:
+        """Store a document to the console."""
+        event.state.next_state()
+        pprint(event.data)
+        self.metrics.number_of_processed_events += 1
+        event.state.next_state(success=True)
+
+    @Output._handle_errors
+    def store_custom(self, event: Event, target: str) -> None:
+        """Store a custom document to the console."""
+        event.state.next_state()
+        pprint(event.data, stream=getattr(sys, target))
+        self.metrics.number_of_processed_events += 1
+        event.state.next_state(success=True)
+
+    def flush(self):
+        """Flush the console output.
+        Do nothing as console output does not have a backlog.
+        """

logprep/ng/connector/console/output.py

@@ -0,0 +1,108 @@
+"""
+DummyOutput
+===========
+
+The Dummy Output Connector can be used to store unmodified documents.
+It only requires the connector type to be configured.
+
+..  code-block:: yaml
+
+Example
+^^^^^^^
+..  code-block:: yaml
+    :linenos:
+
+    output:
+      my_dummy_output:
+        type: dummy_output
+"""
+
+from typing import TYPE_CHECKING, List
+
+from attr import define, field
+from attrs import validators
+
+from logprep.ng.abc.event import Event
+from logprep.ng.abc.output import Output
+from logprep.ng.event.log_event import LogEvent
+
+if TYPE_CHECKING:
+    from logprep.abc.connector import Connector  # pragma: no cover
+
+
+class DummyOutput(Output):
+    """
+    A dummy output that stores unmodified documents unless an exception was raised.
+    """
+
+    @define(kw_only=True)
+    class Config(Output.Config):
+        """Common Configurations"""
+
+        do_nothing: bool = field(default=False)
+        """If set to True, this connector will behave completely neutral and not do anything.
+        Especially counting metrics or storing events."""
+
+        exceptions: List[str] = field(
+            validator=validators.deep_iterable(
+                member_validator=validators.instance_of((str, type(None))),
+                iterable_validator=validators.instance_of(list),
+            ),
+            default=[],
+        )
+        """List of exceptions to raise when storing an event. This is useful
+        for testing purposes. If an exception is raised, the exception is handled
+        by the output decorator.
+        """
+
+    events: list[LogEvent]
+    failed_events: list[LogEvent]
+    setup_called_count: int
+    shut_down_called_count: int
+    _exceptions: list
+
+    __slots__ = [
+        "events",
+        "failed_events",
+        "setup_called_count",
+        "shut_down_called_count",
+        "_exceptions",
+    ]
+
+    def __init__(self, name: str, configuration: "Connector.Config"):
+        super().__init__(name, configuration)
+        self.events = []
+        self.failed_events = []
+        self.shut_down_called_count = 0
+        self._exceptions = configuration.exceptions
+
+    @Output._handle_errors
+    def store(self, event: Event) -> None:
+        """Store the document in the output destination.
+
+        Parameters
+        ----------
+        document : dict
+           Processed log event that will be stored.
+        """
+        if self._config.do_nothing:
+            return
+        event.state.next_state()
+        if self._exceptions:
+            exception = self._exceptions.pop(0)
+            if exception is not None:
+                raise Exception(exception)  # pylint: disable=broad-exception-raised
+        self.events.append(event)
+        event.state.next_state(success=True)
+        self.metrics.number_of_processed_events += 1
+
+    def store_custom(self, event: Event, target: str):  # pylint: disable=unused-argument
+        """Store additional data in a custom location inside the output destination."""
+        self.store(event)
+
+    def shut_down(self):
+        self.shut_down_called_count += 1
+        super().shut_down()
+
+    def flush(self):
+        """Flush not implemented because it has not backlog."""

logprep/ng/connector/dummy/output.py

@@ -0,0 +1,98 @@
+"""
+JsonlOutput
+===========
+
+The JsonlOutput Connector can be used to write processed documents to .jsonl
+files.
+
+Example
+^^^^^^^
+..  code-block:: yaml
+    :linenos:
+
+    output:
+      my_jsonl_output:
+        type: jsonl_output
+        output_file: path/to/output.file
+        output_file_custom: ""
+        output_file_error: ""
+"""
+
+import json
+
+from attrs import define, field, validators
+
+from logprep.ng.abc.event import Event
+from logprep.ng.abc.output import Output
+
+
+class JsonlOutput(Output):
+    """An output that writes the documents it was initialized with to a file.
+
+    Parameters
+    ----------
+    output_path : str
+        The path for the output file.
+    output_path_custom : str
+        The path to store custom
+    output_path_error : str
+        The path to store error
+    """
+
+    @define(kw_only=True)
+    class Config(Output.Config):
+        """Common Configurations"""
+
+        output_file = field(validator=validators.instance_of(str))
+        output_file_custom = field(validator=validators.instance_of(str), default="")
+
+    last_timeout: float
+    events: list[dict]
+    failed_events: list[dict]
+
+    __slots__ = [
+        "last_timeout",
+        "events",
+        "failed_events",
+    ]
+
+    def __init__(self, name: str, configuration: "Output.Config"):
+        super().__init__(name, configuration)
+        self.events = []
+        self.failed_events = []
+
+    def setup(self):
+        super().setup()
+        open(self._config.output_file, "a+", encoding="utf8").close()
+        if self._config.output_file_custom:
+            open(self._config.output_file_custom, "a+", encoding="utf8").close()
+
+    @staticmethod
+    def _write_json(filepath: str, line: dict):
+        """writes processed document to configured file"""
+        with open(filepath, "a+", encoding="utf8") as file:
+            file.write(f"{json.dumps(line)}\n")
+
+    @Output._handle_errors
+    def store(self, event: Event) -> None:
+        """Store the event in the output destination."""
+        event.state.next_state()
+        self.events.append(event.data)
+        JsonlOutput._write_json(self._config.output_file, event.data)
+        self.metrics.number_of_processed_events += 1
+        event.state.next_state(success=True)
+
+    @Output._handle_errors
+    def store_custom(self, event: Event, target: str) -> None:
+        """Store the event in the output destination with a custom target."""
+        event.state.next_state()
+        document = {target: event.data}
+        self.events.append(document)
+
+        if self._config.output_file_custom:
+            JsonlOutput._write_json(self._config.output_file_custom, document)
+        self.metrics.number_of_processed_events += 1
+        event.state.next_state(success=True)
+
+    def flush(self):
+        """Flush is not implemented because it has no backlog."""

logprep/ng/connector/jsonl/__init__.py

@@ -128,6 +128,9 @@ def __str__(self) -> str:
 
         return f"{self.current_state}"
 
+    def __repr__(self) -> str:
+        return self.__str__()
+
     @staticmethod
     def _construct_state_machine() -> dict[EventStateType, list[EventStateType]]:
         """