removed ProcessStrategy as it is implementing additional complexity w…

…ithout multiple Strategies

logprep/abc/processor.py

@@ -1,6 +1,8 @@
 """Abstract module for processors"""
 import copy
+import time
 from abc import abstractmethod
+from functools import reduce
 from logging import DEBUG, Logger
 from multiprocessing import current_process
 from pathlib import Path
@@ -16,7 +18,6 @@
     ProcessingCriticalError,
     ProcessingWarning,
 )
-from logprep.processor.processor_strategy import SpecificGenericProcessStrategy
 from logprep.util import getter
 from logprep.util.helper import (
     add_and_overwrite,
@@ -120,7 +121,6 @@ def update_mean_processing_time_per_event(self, new_sample):
 
     def __init__(self, name: str, configuration: "Processor.Config", logger: Logger):
         super().__init__(name, configuration, logger)
-        self._strategy = SpecificGenericProcessStrategy(self._config.apply_multiple_times)
         self.metric_labels, specific_tree_labels, generic_tree_labels = self._create_metric_labels()
         self._specific_tree = RuleTree(
             config_path=self._config.tree_config, metric_labels=specific_tree_labels
@@ -192,13 +192,31 @@ def process(self, event: dict):
         """
         if self._logger.isEnabledFor(DEBUG):  # pragma: no cover
             self._logger.debug(f"{self.describe()} processing event {event}")
-        self._strategy.process(
-            event,
-            generic_tree=self._generic_tree,
-            specific_tree=self._specific_tree,
-            callback=self._apply_rules_wrapper,
-            processor_metrics=self.metrics,
-        )
+
+        self.metrics.number_of_processed_events += 1
+        self._process_rule_tree(event, self._specific_tree)
+        self._process_rule_tree(event, self._generic_tree)
+
+    def _process_rule_tree(self, event: dict, tree: "RuleTree"):
+        applied_rules = set()
+
+        def _process_rule(event, rule):
+            begin = time.time()
+            self._apply_rules_wrapper(event, rule)
+            processing_time = time.time() - begin
+            rule.metrics._number_of_matches += 1
+            rule.metrics.update_mean_processing_time(processing_time)
+            self.metrics.update_mean_processing_time_per_event(processing_time)
+            applied_rules.add(rule)
+            return event
+
+        if self._config.apply_multiple_times:
+            matching_rules = tree.get_matching_rules(event)
+            while matching_rules:
+                reduce(_process_rule, (event, *matching_rules))
+                matching_rules = set(tree.get_matching_rules(event)).difference(applied_rules)
+        else:
+            reduce(_process_rule, (event, *tree.get_matching_rules(event)))
 
     def _apply_rules_wrapper(self, event, rule):
         try:

tests/performance/test_grok.py

@@ -0,0 +1,79 @@
+# pylint: disable=missing-docstring
+# pylint: disable=protected-access
+import timeit
+
+setup_import = """
+from unittest import mock
+
+from logprep.factory import Factory
+from logprep.processor.grokker.processor import Grokker
+from logprep.processor.grokker.rule import GrokkerRule
+"""
+setup_grokker = """
+rule = GrokkerRule._create_from_dict(rule)
+grokker_config = {
+    "mygrokker": {
+        "type": "grokker",
+        "specific_rules": [],
+        "generic_rules": [],
+    }
+}
+mock_logger = mock.MagicMock()
+grokker: Grokker = Factory.create(grokker_config, mock_logger)
+grokker._specific_tree.add_rule(rule)
+grokker.setup()
+"""
+run = """
+grokker.process(document)
+    """
+
+simple_grok_pattern = """
+document = {"message": "2020-07-16T19:20:30.45+01:00 DEBUG This is a sample log"}
+rule = {
+    "filter": "message",
+    "grokker": {
+        "mapping": {
+            "message": "%{TIMESTAMP_ISO8601:@timestamp} %{LOGLEVEL:logLevel} %{GREEDYDATA:logMessage}"
+        }
+    },
+}
+"""
+
+linux_syslogline_pattern = """
+document = {"message": 'Oct  7 09:21:35 dev-machine c6182927c772[1115]: logger=infra.usagestats t=2023-10-07T09:21:35.676177822Z level=info msg="Usage stats are ready to report"'}
+rule = {
+    "filter": "message",
+    "grokker": {
+        "mapping": {
+            "message": "%{SYSLOGLINE}"
+        }
+    },
+}
+"""
+
+linux_syslogline_5424_pattern = """
+document = {"message": 'Oct  7 09:21:35 dev-machine c6182927c772[1115]: logger=infra.usagestats t=2023-10-07T09:21:35.676177822Z level=info msg="Usage stats are ready to report"'}
+rule = {
+    "filter": "message",
+    "grokker": {
+        "mapping": {
+            "message": "%{SYSLOG5424LINE}"
+        }
+    },
+}
+"""
+
+
+def main():
+    testcases = [
+        simple_grok_pattern,
+        linux_syslogline_pattern,
+        linux_syslogline_5424_pattern,
+    ]
+
+    for case in testcases:
+        print(timeit.timeit(run, number=100000, setup=setup_import + case + setup_grokker))
+
+
+if __name__ == "__main__":
+    main()

tests/unit/processor/base.py

@@ -16,7 +16,6 @@
 from logprep.factory import Factory
 from logprep.framework.rule_tree.rule_tree import RuleTree
 from logprep.processor.base.exceptions import ProcessingWarning
-from logprep.processor.processor_strategy import ProcessStrategy
 from logprep.util.helper import camel_to_snake
 from logprep.util.json_handling import list_json_files_in_directory
 from logprep.util.time_measurement import TimeMeasurement
@@ -228,20 +227,6 @@ def test_rules_returns_all_specific_and_generic_rules(self):
         object_rules_count = len(self.object.rules)
         assert all_rules_count == object_rules_count
 
-    def test_process_strategy_returns_strategy_object(self):
-        assert isinstance(self.object._strategy, ProcessStrategy)
-
-    def test_process_calls_strategy(self):
-        """
-        This test method needs to be overwritten in your ProcessorTests
-        if your processor uses another strategy
-        """
-        with mock.patch(
-            "logprep.processor.processor_strategy.SpecificGenericProcessStrategy.process"
-        ) as mock_strategy_process:
-            self.object.process({})
-            mock_strategy_process.assert_called()
-
     def test_process_is_measured(self):
         TimeMeasurement.TIME_MEASUREMENT_ENABLED = True
         TimeMeasurement.APPEND_TO_EVENT = True