make process method always return a result object

.github/workflows/testing.yml

@@ -26,6 +26,7 @@ jobs:
         run: |
           pip install --upgrade pip wheel
           pip install .[dev]
+          pip list
       - name: Perform ${{ matrix.test-type }} test
         env:
           PYTEST_ADDOPTS: "--color=yes"

CHANGELOG.md

@@ -4,6 +4,10 @@
 ### Breaking
 ### Features
 ### Improvements
+
+* a result object was added which is returned by every processor
+  * includes the processor name, generated extra_data, warnings and errors
+
 ### Bugfix
 
 ## 13.0.0

logprep/abc/processor.py

@@ -3,7 +3,7 @@
 import logging
 from abc import abstractmethod
 from pathlib import Path
-from typing import TYPE_CHECKING, List, Optional, Tuple
+from typing import TYPE_CHECKING, List, Optional
 
 from attr import define, field, validators
 
@@ -13,6 +13,7 @@
 from logprep.processor.base.exceptions import (
     FieldExistsWarning,
     ProcessingCriticalError,
+    ProcessingError,
     ProcessingWarning,
 )
 from logprep.util import getter
@@ -30,6 +31,36 @@
 logger = logging.getLogger("Processor")
 
 
+@define(kw_only=True)
+class ProcessorResult:
+    """Result object to be returned by every processor. It contains all extra_data and errors."""
+
+    name: str = field(validator=validators.instance_of(str))
+    data: list = field(validator=validators.instance_of(list), factory=list)
+    errors: list = field(
+        validator=validators.deep_iterable(
+            member_validator=validators.instance_of((ProcessingError, ProcessingWarning)),
+            iterable_validator=validators.instance_of(list),
+        ),
+        factory=list,
+    )
+
+    def __contains__(self, error_class):
+        return any(isinstance(item, error_class) for item in self.errors)
+
+    def get_warning_string(self):
+        """creates a string containing the warnings"""
+        return ", ".join(
+            [error.args[0] for error in self.errors if isinstance(error, ProcessingWarning)]
+        )
+
+    def get_error_string(self):
+        """creates a string containing the errors"""
+        return ", ".join(
+            [error.args[0] for error in self.errors if isinstance(error, ProcessingError)]
+        )
+
+
 class Processor(Component):
     """Abstract Processor Class to define the Interface"""
 
@@ -76,16 +107,16 @@ class Config(Component.Config):
         "_event",
         "_specific_tree",
         "_generic_tree",
-        "_extra_data",
+        "result",
     ]
 
     rule_class: "Rule"
     has_custom_tests: bool
     _event: dict
     _specific_tree: RuleTree
     _generic_tree: RuleTree
-    _extra_data: List[Tuple[dict, Tuple[dict]]]
     _strategy = None
+    result: ProcessorResult
 
     def __init__(self, name: str, configuration: "Processor.Config"):
         super().__init__(name, configuration)
@@ -104,7 +135,7 @@ def __init__(self, name: str, configuration: "Processor.Config"):
             specific_rules_targets=self._config.specific_rules,
         )
         self.has_custom_tests = False
-        self._extra_data = []
+        self.result = ProcessorResult(name=self.name)
 
     @property
     def _specific_rules(self):
@@ -156,11 +187,11 @@ def process(self, event: dict):
            A dictionary representing a log event.
 
         """
-        self._extra_data.clear()
+        self.result = ProcessorResult(name=self.name)
         logger.debug(f"{self.describe()} processing event {event}")
         self._process_rule_tree(event, self._specific_tree)
         self._process_rule_tree(event, self._generic_tree)
-        return self._extra_data if self._extra_data else None
+        return self.result
 
     def _process_rule_tree(self, event: dict, tree: "RuleTree"):
         applied_rules = set()
@@ -195,9 +226,9 @@ def _apply_rules_wrapper(self, event: dict, rule: "Rule"):
         except ProcessingWarning as error:
             self._handle_warning_error(event, rule, error)
         except ProcessingCriticalError as error:
-            raise error  # is needed to prevent wrapping it in itself
+            self.result.errors.append(error)  # is needed to prevent wrapping it in itself
         except BaseException as error:
-            raise ProcessingCriticalError(str(error), rule, event) from error
+            self.result.errors.append(ProcessingCriticalError(str(error), rule, event))
         if not hasattr(rule, "delete_source_fields"):
             return
         if rule.delete_source_fields:
@@ -285,9 +316,9 @@ def _handle_warning_error(self, event, rule, error, failure_tags=None):
         else:
             add_and_overwrite(event, "tags", sorted(list({*tags, *failure_tags})))
         if isinstance(error, ProcessingWarning):
-            logger.warning(str(error))
+            self.result.errors.append(error)
         else:
-            logger.warning(str(ProcessingWarning(str(error), rule, event)))
+            self.result.errors.append(ProcessingWarning(str(error), rule, event))
 
     def _has_missing_values(self, event, rule, source_field_dict):
         missing_fields = list(

logprep/framework/pipeline.py

@@ -6,6 +6,7 @@
 """
 
 import copy
+import itertools
 import logging
 import logging.handlers
 import multiprocessing
@@ -37,14 +38,32 @@
     Output,
     OutputWarning,
 )
-from logprep.abc.processor import Processor
+from logprep.abc.processor import Processor, ProcessorResult
 from logprep.factory import Factory
 from logprep.metrics.metrics import HistogramMetric, Metric
-from logprep.processor.base.exceptions import ProcessingCriticalError, ProcessingWarning
+from logprep.processor.base.exceptions import ProcessingError, ProcessingWarning
 from logprep.util.configuration import Configuration
 from logprep.util.pipeline_profiler import PipelineProfiler
 
 
+@attrs.define(kw_only=True)
+class PipelineResult:
+    """Result object to be returned after processing the event.
+    It contains all generated data and includes errors and warnings."""
+
+    results: List[ProcessorResult] = attrs.field(
+        validator=[
+            attrs.validators.instance_of(list),
+            attrs.validators.deep_iterable(
+                member_validator=attrs.validators.instance_of(ProcessorResult)
+            ),
+        ]
+    )
+
+    def __iter__(self):
+        return iter(self.results)
+
+
 def _handle_pipeline_error(func):
     def _inner(self: "Pipeline") -> Any:
         try:
@@ -218,68 +237,71 @@ def run(self) -> None:  # pylint: disable=method-hidden
     def process_pipeline(self) -> Tuple[dict, list]:
         """Retrieve next event, process event with full pipeline and store or return results"""
         Component.run_pending_tasks()
-        extra_outputs = []
-        event = None
-        try:
-            event = self._get_event()
-        except CriticalInputParsingError as error:
-            input_data = error.raw_input
-            if isinstance(input_data, bytes):
-                input_data = input_data.decode("utf8")
-            error_event = self._encoder.encode({"invalid_json": input_data})
-            self._store_failed_event(error, "", error_event)
-            self.logger.error(f"{error}, event was written to error output")
-        if event:
-            extra_outputs = self.process_event(event)
-        if event and self._output:
-            self._store_event(event)
-        return event, extra_outputs
+
+        event = self._get_event()
+        event_received = copy.deepcopy(event)
+        if not event:
+            return None, None
+        result: PipelineResult = self.process_event(event)
+        for processor_result in result:
+            if not processor_result.errors:
+                continue
+            if ProcessingWarning in processor_result:
+                self.logger.warning(processor_result.get_warning_string())
+            if ProcessingError in processor_result:
+                self.logger.error(processor_result.get_error_string())
+                if self._output:
+                    self._store_failed_event(processor_result.errors, event_received, event)
+                # pipeline is aborted on processing error
+                return event, result
+        if self._output:
+            result_data = [res.data for res in result if res.data]
+            result_data = itertools.chain(*result_data)
+            if result_data:
+                self._store_extra_data(result_data)
+            if event:
+                self._store_event(event)
+        return event, result
 
     def _store_event(self, event: dict) -> None:
         for output_name, output in self._output.items():
             if output.default:
                 output.store(event)
                 self.logger.debug(f"Stored output in {output_name}")
 
-    def _store_failed_event(self, error, event, event_received):
+    def _store_failed_event(self, error, event_received, event):
         for _, output in self._output.items():
             if output.default:
-                output.store_failed(str(error), self._decoder.decode(event_received), event)
+                output.store_failed(str(error), event_received, event)
 
     def _get_event(self) -> dict:
-        event, non_critical_error_msg = self._input.get_next(self._timeout)
-        if non_critical_error_msg and self._output:
-            for _, output in self._output.items():
-                if output.default:
-                    output.store_failed(non_critical_error_msg, event, None)
-        return event
+        try:
+            event, non_critical_error_msg = self._input.get_next(self._timeout)
+            if non_critical_error_msg and self._output:
+                self._store_failed_event(non_critical_error_msg, event, None)
+            return event
+        except CriticalInputParsingError as error:
+            input_data = error.raw_input
+            if isinstance(input_data, bytes):
+                input_data = input_data.decode("utf8")
+            self._store_failed_event(error, {"invalid_json": input_data}, "")
 
     @Metric.measure_time()
     def process_event(self, event: dict):
         """process all processors for one event"""
-
-        event_received = self._encoder.encode(event)
-        extra_outputs = []
+        results = []
         for processor in self._pipeline:
-            try:
-                if extra_data := processor.process(event):
-                    if self._output:
-                        self._store_extra_data(extra_data)
-                    extra_outputs.append(extra_data)
-            except ProcessingWarning as error:
-                self.logger.warning(str(error))
-            except ProcessingCriticalError as error:
-                self.logger.error(str(error))
-                if self._output:
-                    self._store_failed_event(error, copy.deepcopy(event), event_received)
-                    event.clear()
+            result: ProcessorResult = processor.process(event)
+            results.append(result)
+            if ProcessingError in result:
+                event.clear()
             if not event:
                 break
-        return extra_outputs
+        return PipelineResult(results=results)
 
-    def _store_extra_data(self, extra_data: List[tuple]) -> None:
+    def _store_extra_data(self, result_data: List) -> None:
         self.logger.debug("Storing extra data")
-        for document, outputs in extra_data:
+        for document, outputs in result_data:
             for output in outputs:
                 for output_name, target in output.items():
                     self._output[output_name].store_custom(document, target)

logprep/processor/pre_detector/processor.py

@@ -38,7 +38,7 @@
 from logprep.abc.processor import Processor
 from logprep.processor.pre_detector.ip_alerter import IPAlerter
 from logprep.processor.pre_detector.rule import PreDetectorRule
-from logprep.util.helper import get_dotted_field_value, add_field_to
+from logprep.util.helper import add_field_to, get_dotted_field_value
 from logprep.util.time import TimeParser
 
 
@@ -98,7 +98,7 @@ def _apply_rules(self, event, rule):
             and not self._ip_alerter.is_in_alerts_list(rule, event)
         ):
             self._get_detection_result(event, rule)
-        for detection, _ in self._extra_data:
+        for detection, _ in self.result.data:
             detection["creation_timestamp"] = TimeParser.now().isoformat()
             timestamp = get_dotted_field_value(event, "@timestamp")
             if timestamp is not None:
@@ -111,7 +111,7 @@ def _get_detection_result(self, event: dict, rule: PreDetectorRule):
             add_field_to(event, "pre_detection_id", pre_detection_id)
 
         detection_result = self._generate_detection_result(pre_detection_id, event, rule)
-        self._extra_data.append((detection_result, self._config.outputs))
+        self.result.data.append((detection_result, self._config.outputs))
 
     @staticmethod
     def _generate_detection_result(pre_detection_id: str, event: dict, rule: PreDetectorRule):

logprep/processor/pseudonymizer/processor.py

@@ -266,7 +266,7 @@ def _apply_rules(self, event: dict, rule: PseudonymizerRule):
                 field_value = self._pseudonymize_field(rule, dotted_field, regex, field_value)
             _ = add_field_to(event, dotted_field, field_value, overwrite_output_field=True)
         if "@timestamp" in event:
-            for pseudonym, _ in self._extra_data:
+            for pseudonym, _ in self.result.data:
                 pseudonym["@timestamp"] = event["@timestamp"]
         self._update_cache_metrics()
 
@@ -297,8 +297,8 @@ def _pseudonymize_string(self, value: str) -> str:
             return value
         pseudonym_dict = self._get_pseudonym_dict_cached(value)
         extra = (pseudonym_dict, self._config.outputs)
-        if extra not in self._extra_data:
-            self._extra_data.append(extra)
+        if extra not in self.result.data:
+            self.result.data.append(extra)
         return self._wrap_hash(pseudonym_dict["pseudonym"])
 
     def _pseudonymize(self, value):

logprep/processor/selective_extractor/processor.py

@@ -66,4 +66,4 @@ def _apply_rules(self, event: dict, rule: SelectiveExtractorRule):
             filtered_event = {}
             for field, content in flattened_fields.items():
                 add_field_to(filtered_event, field, content)
-            self._extra_data.append((filtered_event, rule.outputs))
+            self.result.data.append((filtered_event, rule.outputs))