fix failed event metric

* move `number_of_failed_events` metric from connector to component
* count this metric for input and output criticals in pipeline.py enqueue_error
* count this metric for processing criticals in logprep.abc.processor.Processor.apply_rules_wrapper

logprep/abc/component.py

@@ -14,7 +14,7 @@
 from attrs import asdict
 from schedule import Scheduler
 
-from logprep.metrics.metrics import Metric
+from logprep.metrics.metrics import CounterMetric, Metric
 from logprep.util.defaults import DEFAULT_HEALTH_TIMEOUT, EXITCODES
 from logprep.util.helper import camel_to_snake
 
@@ -47,6 +47,14 @@ class Metrics:
 
         _labels: dict
 
+        number_of_failed_events: CounterMetric = field(
+            factory=lambda: CounterMetric(
+                description="Number of failed events",
+                name="number_of_failed_events",
+            )
+        )
+        """Number of failed events"""
+
         def __attrs_post_init__(self):
             for attribute in asdict(self):
                 attribute = getattr(self, attribute)

logprep/abc/connector.py

@@ -21,14 +21,6 @@ class Metrics(Component.Metrics):
         )
         """Number of successful events"""
 
-        number_of_failed_events: CounterMetric = field(
-            factory=lambda: CounterMetric(
-                description="Number of failed events",
-                name="number_of_failed_events",
-            )
-        )
-        """Number of failed events"""
-
         processing_time_per_event: HistogramMetric = field(
             factory=lambda: HistogramMetric(
                 description="Time in seconds that it took to store an event",

logprep/abc/input.py

@@ -27,6 +27,7 @@ class InputError(LogprepException):
 
     def __init__(self, input_connector: "Input", message: str) -> None:
         input_connector.metrics.number_of_errors += 1
+        self.input = input_connector
         super().__init__(f"{self.__class__.__name__} in {input_connector.describe()}: {message}")
 
 

logprep/abc/output.py

@@ -17,6 +17,7 @@ class OutputError(LogprepException):
 
     def __init__(self, output: "Output", message: str) -> None:
         output.metrics.number_of_errors += 1
+        self.output = output
         super().__init__(f"{self.__class__.__name__} in {output.describe()}: {message}")
 
 

logprep/abc/processor.py

@@ -272,6 +272,7 @@ def _apply_rules_wrapper(self, event: dict, rule: "Rule"):
         except ProcessingWarning as error:
             self._handle_warning_error(event, rule, error)
         except ProcessingCriticalError as error:
+            self.metrics.number_of_failed_events += 1
             self.result.errors.append(error)  # is needed to prevent wrapping it in itself
             event.clear()
         except Exception as error:  # pylint: disable=broad-except

logprep/framework/pipeline.py

@@ -371,6 +371,7 @@ def enqueue_error(
                     "errors": ", ".join((str(error.message) for error in errors)),
                 }
             case CriticalInputError():
+                item.input.metrics.number_of_failed_events += 1
                 event = {"event": str(item.raw_input), "errors": str(item.message)}
             case list():
                 event = [{"event": str(i), "errors": "Unknown error"} for i in item]
@@ -400,14 +401,21 @@ def _get_output_error_event(self, item: CriticalOutputError) -> dict:
                 event = [
                     {"event": str(i["event"]), "errors": str(i["errors"])} for i in item.raw_input
                 ]
+                item.output.metrics.number_of_failed_events += len(event)
                 return event
             case CriticalOutputError({"errors": error, "event": event}):
+                item.output.metrics.number_of_failed_events += 1
                 return {"event": str(event), "errors": str(error)}
             case CriticalOutputError(raw_input) if isinstance(raw_input, dict):
+                item.output.metrics.number_of_failed_events += 1
                 return {"event": str(raw_input), "errors": str(item.message)}
             case CriticalOutputError(raw_input) if isinstance(raw_input, (list, tuple)):
-                return [{"event": str(i), "errors": str(item.message)} for i in raw_input]
+                event = [{"event": str(i), "errors": str(item.message)} for i in raw_input]
+                item.output.metrics.number_of_failed_events += len(event)
+                return event
             case CriticalOutputError(raw_input) if isinstance(raw_input, (str, bytes)):
+                item.output.metrics.number_of_failed_events += 1
                 return {"event": str(raw_input), "errors": str(item.message)}
             case _:
+                item.output.metrics.number_of_failed_events += 1
                 return {"event": str(item.raw_input), "errors": str(item.message)}

tests/unit/component/base.py

@@ -31,6 +31,10 @@ class BaseComponentTestCase(ABC):
 
     metric_attributes: dict
 
+    expected_metrics = [
+        "logprep_number_of_failed_events",
+    ]
+
     def setup_method(self) -> None:
         config = {"Test Instance Name": self.CONFIG}
         self.object = Factory.create(configuration=config)

tests/unit/processor/amides/test_amides.py

@@ -25,6 +25,7 @@ class TestAmides(BaseProcessorTestCase):
     }
 
     expected_metrics = [
+        "logprep_number_of_failed_events",
         "logprep_amides_total_cmdlines",
         "logprep_amides_new_results",
         "logprep_amides_cached_results",

tests/unit/processor/domain_resolver/test_domain_resolver.py

@@ -34,6 +34,7 @@ class TestDomainResolver(BaseProcessorTestCase):
 
     expected_metrics = [
         "logprep_domain_resolver_total_urls",
+        "logprep_number_of_failed_events",
         "logprep_domain_resolver_resolved_new",
         "logprep_domain_resolver_resolved_cached",
         "logprep_domain_resolver_timeouts",

tests/unit/processor/pseudonymizer/test_pseudonymizer.py

@@ -715,6 +715,7 @@ class TestPseudonymizer(BaseProcessorTestCase):
     }
 
     expected_metrics = [
+        "logprep_number_of_failed_events",
         "logprep_pseudonymizer_pseudonymized_urls",
         "logprep_pseudonymizer_new_results",
         "logprep_pseudonymizer_cached_results",