WIP

logprep/connector/console/output.py

@@ -33,4 +33,5 @@ def store_custom(self, document: dict, target: str):
         pprint(document, stream=getattr(sys, target))
 
     def store_failed(self, error_message: str, document_received: dict, document_processed: dict):
+        self.metrics.number_of_failed_events += 1
         pprint(f"{error_message}: {document_received}, {document_processed}", stream=sys.stderr)

logprep/framework/pipeline.py

@@ -67,7 +67,6 @@ def _init_timer(self, period: float):
 
     def setup(self, print_processed_period: float, lock: Lock):
         """Setup shared counter for multiprocessing pipeline."""
-
         self._init_timer(print_processed_period)
         self._lock = lock
 
@@ -207,8 +206,9 @@ def _event_version_information(self) -> dict:
         }
 
     @property
-    def _metric_labels(self) -> dict:
-        return {"pipeline": self._process_name}
+    def metric_labels(self) -> dict:
+        """Return the metric labels for this component."""
+        return {"component": "pipeline"}
 
     @cached_property
     def metrics(self) -> Metrics:
@@ -218,7 +218,7 @@ def metrics(self) -> Metrics:
         return self.Metrics(
             input=self._input.metrics,
             output=[self._output.get(output).metrics for output in self._output],
-            labels=self._metric_labels,
+            labels=self.metric_labels,
         )
 
     @cached_property

logprep/run_logprep.py

@@ -101,7 +101,6 @@ def _run_logprep(arguments, logger: logging.Logger):
         logger.critical(f"A critical error occurred: {error}")
         if runner:
             runner.stop()
-        raise error
         sys.exit(1)
     # pylint: enable=broad-except