fix metrics tests

logprep/abc/component.py

@@ -59,7 +59,7 @@ def __attrs_post_init__(self):
                 if isinstance(attribute, Metric):
                     attribute.labels = self._labels
                     attribute.target = self._processing_time_per_event_target
-                    attribute.tracker = attribute.init_tracker()
+                    attribute.init_tracker()
 
     # __dict__ is added to support functools.cached_property
     __slots__ = ["name", "_logger", "_config", "__dict__"]

logprep/abc/processor.py

@@ -255,11 +255,11 @@ def load_rules(self, specific_rules_targets: List[str], generic_rules_targets: L
         specific_rules_targets = self.resolve_directories(specific_rules_targets)
         generic_rules_targets = self.resolve_directories(generic_rules_targets)
         for specific_rules_target in specific_rules_targets:
-            rules = self.rule_class.create_rules_from_target(specific_rules_target, processor=self)
+            rules = self.rule_class.create_rules_from_target(specific_rules_target)
             for rule in rules:
                 self._specific_tree.add_rule(rule, self._logger)
         for generic_rules_target in generic_rules_targets:
-            rules = self.rule_class.create_rules_from_target(generic_rules_target, processor=self)
+            rules = self.rule_class.create_rules_from_target(generic_rules_target)
             for rule in rules:
                 self._generic_tree.add_rule(rule, self._logger)
 

logprep/framework/pipeline.py

@@ -352,7 +352,7 @@ def process_event(self, event: dict):
             - create Grafana Dashboards
             - add pipelinemanager metrics (pipeline restarts)
             - clean up PrometheusExporter ("remove stale metric files" stil needed?)
-            - 
+            - add Kafka librdkafka metrics
         """
 
         event_received = self._encoder.encode(event)

logprep/metrics/metrics.py

@@ -5,6 +5,8 @@
 from attr import asdict, define, field, validators
 from prometheus_client import Counter, Histogram
 
+LOGPREP_REGISTRY = None  # to inject a custom registry
+
 
 def is_public(attribute, _):
     """If an attribute name starts with an underscore it is considered private"""
@@ -60,18 +62,18 @@ def init_tracker(self):
                 name=f"{self._prefix}{self.name}",
                 documentation=self.description,
                 labelnames=self.labels.keys(),
-                registry=None,
+                registry=LOGPREP_REGISTRY,
             )
         if isinstance(self, HistogramMetric):
             tracker = Histogram(
                 name=f"{self._prefix}{self.name}",
                 documentation=self.description,
                 labelnames=self.labels.keys(),
                 buckets=(0.000001, 0.00001, 0.0001, 0.001, 0.01, 0.1, 1),
-                registry=None,
+                registry=LOGPREP_REGISTRY,
             )
         tracker.labels(**self.labels)
-        return tracker
+        self.tracker = tracker
 
     @abstractmethod
     def __add__(self, other):

logprep/processor/base/rule.py

@@ -276,17 +276,17 @@ def lucene_filter(self):
     # pylint: enable=C0111
 
     @classmethod
-    def create_rules_from_target(cls, rule_target: str, processor: Processor) -> list:
+    def create_rules_from_target(cls, rule_target: str) -> list:
         """Create a rule from a file."""
         if isinstance(rule_target, dict):
-            return [cls._create_from_dict(rule_target, processor.name)]
+            return [cls._create_from_dict(rule_target)]
         content = GetterFactory.from_string(rule_target).get()
         try:
             rule_data = json.loads(content)
         except ValueError:
             rule_data = yaml.load_all(content)
         try:
-            rules = [cls._create_from_dict(rule, processor.name) for rule in rule_data]
+            rules = [cls._create_from_dict(rule) for rule in rule_data]
         except InvalidRuleDefinitionError as error:
             raise InvalidRuleDefinitionError(f"{rule_target}: {error}") from error
         if len(rules) == 0:
@@ -302,7 +302,7 @@ def normalize_rule_dict(cls, rule: dict) -> None:
         """
 
     @classmethod
-    def _create_from_dict(cls, rule: dict, processor_name: str) -> "Rule":
+    def _create_from_dict(cls, rule: dict) -> "Rule":
         cls.normalize_rule_dict(rule)
         filter_expression = Rule._create_filter_expression(rule)
         cls.rule_type = camel_to_snake(cls.__name__.replace("Rule", ""))
@@ -319,7 +319,7 @@ def _create_from_dict(cls, rule: dict, processor_name: str) -> "Rule":
             if special_field_value is not None:
                 config.update({special_field: special_field_value})
         config = cls.Config(**config)
-        return cls(filter_expression, config, processor_name)
+        return cls(filter_expression, config, cls.rule_type)
 
     @staticmethod
     def _check_rule_validity(

logprep/processor/normalizer/rule.py

@@ -1007,7 +1007,7 @@ def timestamps(self) -> dict:
     # pylint: enable=C0111
 
     @staticmethod
-    def _create_from_dict(rule: dict, processor_name: str) -> "NormalizerRule":
+    def _create_from_dict(rule: dict) -> "NormalizerRule":
         NormalizerRule._check_rule_validity(rule, "normalize")
         NormalizerRule._check_if_normalization_valid(rule)
 

logprep/util/configuration.py

@@ -7,11 +7,9 @@
 from pathlib import Path
 from typing import List
 
-from attrs import define, field
 from colorama import Fore
 from ruamel.yaml.scanner import ScannerError
 
-from logprep.abc.component import Component
 from logprep.abc.getter import Getter
 from logprep.abc.processor import Processor
 from logprep.factory import Factory

tests/unit/metrics/test_metrics.py

@@ -1,42 +1,44 @@
 # pylint: disable=missing-docstring
 # pylint: disable=no-self-use
 # pylint: disable=protected-access
-from typing import List
 
-import numpy as np
-from prometheus_client import REGISTRY
-from prometheus_client.registry import Collector
+from prometheus_client import CollectorRegistry, Counter, generate_latest
 
-from logprep.metrics.metrics import CounterMetric, MetricType
+import logprep.metrics.metrics as metrics
+from logprep.metrics.metrics import CounterMetric
+
+custom_registry = CollectorRegistry()
+
+metrics.LOGPREP_REGISTRY = custom_registry
 
 
 class TestsMetrics:
-    def test_converts_enum_to_prometheus_metric(self):
+    def test_init_tracker_creates_metric(self):
         metric = CounterMetric(
             name="testmetric",
-            type=MetricType.COUNTER,
             description="empty description",
             labels={"A": "a"},
         )
-        assert issubclass(metric.type, Collector)
+        metric.init_tracker()
+        assert isinstance(metric.tracker, Counter)
 
     def test_counter_metric_sets_labels(self):
         metric = CounterMetric(
-            type=MetricType.COUNTER,
             name="bla",
             description="empty description",
             labels={"pipeline": "pipeline-1"},
         )
+        metric.init_tracker()
         assert metric.tracker._labelnames == ("pipeline",)
         assert ("pipeline-1",) in metric.tracker._metrics
 
     def test_counter_metric_increments_correctly(self):
         metric = CounterMetric(
-            type=MetricType.COUNTER,
             name="bla",
             description="empty description",
             labels={"pipeline": "1"},
         )
+        metric.init_tracker()
         metric += 1
-        assert list(REGISTRY.collect())[-1].samples[-2].value == 1
-        assert metric is not None
+        metric_output = generate_latest(custom_registry).decode("utf-8")
+        assert 'logprep_bla_total{pipeline="1"} 1.0' in metric_output

tests/unit/processor/base.py

@@ -2,6 +2,9 @@
 # pylint: disable=protected-access
 
 import json
+import os
+import shutil
+import tempfile
 from copy import deepcopy
 from logging import getLogger
 from pathlib import Path
@@ -10,11 +13,20 @@
 import pytest
 import requests
 import responses
+from prometheus_client import (
+    REGISTRY,
+    CollectorRegistry,
+    generate_latest,
+    multiprocess,
+    values,
+)
 from ruamel.yaml import YAML
 
+from logprep import metrics
 from logprep.abc.processor import Processor
 from logprep.factory import Factory
 from logprep.framework.rule_tree.rule_tree import RuleTree
+from logprep.metrics.metrics import Metric
 from logprep.processor.base.exceptions import ProcessingWarning
 from logprep.util.helper import camel_to_snake
 from logprep.util.json_handling import list_json_files_in_directory
@@ -86,6 +98,8 @@ def setup_method(self) -> None:
         """
         TimeMeasurement.TIME_MEASUREMENT_ENABLED = False
         TimeMeasurement.APPEND_TO_EVENT = False
+        self.registry = CollectorRegistry()
+        metrics.LOGPREP_REGISTRY = self.registry
         self.patchers = []
         for name, kwargs in self.mocks.items():
             patcher = mock.patch(name, **kwargs)
@@ -106,14 +120,14 @@ def test_is_a_processor_implementation(self):
         assert isinstance(self.object, Processor)
 
     def test_process(self):
-        assert self.object.metrics.number_of_processed_events == 0
+        before = generate_latest(self.registry)
         document = {
             "event_id": "1234",
             "message": "user root logged in",
         }
         count = self.object.metrics.number_of_processed_events
         self.object.process(document)
-
+        after = generate_latest(self.registry)
         assert self.object.metrics.number_of_processed_events == count + 1
 
     def test_generic_specific_rule_trees(self):

tests/unit/util/test_configuration.py

@@ -12,7 +12,6 @@
 
 from logprep.util.configuration import (
     Configuration,
-    IncalidMetricsConfigurationError,
     InvalidConfigurationError,
     InvalidConfigurationErrors,
     InvalidInputConnectorConfigurationError,