Merge branch 'main' into dev-revise-add-field-to-error-handling

CHANGELOG.md

@@ -4,8 +4,13 @@
 ### Breaking
 ### Features
 ### Improvements
+* replace `BaseException` with `Exception` for custom errors
 ### Bugfix
 
+- fix `confluent_kafka.store_offsets` if `last_valid_record` is `None`, can happen if a rebalancing happens
+  before the first message was pulled.
+- fix pseudonymizer cache metrics not updated
+
 ## 14.0.0
 ### Breaking
 
@@ -53,7 +58,7 @@
 
 * fixes a bug not increasing but decreasing timeout throttle factor of ThrottlingQueue
 * handle DecodeError and unexpected Exceptions on requests in `http_input` separately
-* fixes unbound local error in http input connector 
+* fixes unbound local error in http input connector
 
 ## 13.1.1
 ### Improvements
@@ -97,7 +102,7 @@
 
 ### Bugfix
 
-* This release limits the mysql-connector-python dependency to have version less the 9 
+* This release limits the mysql-connector-python dependency to have version less the 9
 
 ## 13.0.0
 

logprep/abc/processor.py

@@ -375,7 +375,7 @@ def _has_missing_values(self, event, rule, source_field_dict):
         if missing_fields:
             if rule.ignore_missing_fields:
                 return True
-            error = BaseException(f"{self.name}: no value for fields: {missing_fields}")
+            error = Exception(f"{self.name}: no value for fields: {missing_fields}")
             self._handle_warning_error(event, rule, error)
             return True
         return False

logprep/connector/confluent_kafka/input.py

@@ -473,6 +473,9 @@ def batch_finished_callback(self) -> None:
         """
         if self._enable_auto_offset_store:
             return
+        # in case the ConfluentKafkaInput._revoke_callback is triggered before the first message was polled
+        if not self._last_valid_record:
+            return
         try:
             self._consumer.store_offsets(message=self._last_valid_record)
         except KafkaException as error:

logprep/connector/dummy/input.py

@@ -4,7 +4,7 @@
 
 A dummy input that returns the documents it was initialized with.
 
-If a "document" is derived from BaseException, that exception will be thrown instead of
+If a "document" is derived from Exception, that exception will be thrown instead of
 returning a document. The exception will be removed and subsequent calls may return documents or
 throw other exceptions in the given order.
 
@@ -36,7 +36,7 @@ class DummyInput(Input):
     class Config(Input.Config):
         """DummyInput specific configuration"""
 
-        documents: List[Union[dict, type, BaseException]]
+        documents: List[Union[dict, type, Exception]]
         """A list of documents that should be returned."""
         repeat_documents: Optional[str] = field(
             validator=validators.instance_of(bool), default=False
@@ -57,6 +57,6 @@ def _get_event(self, timeout: float) -> tuple:
 
         document = self._documents.pop(0)
 
-        if (document.__class__ == type) and issubclass(document, BaseException):
+        if (document.__class__ == type) and issubclass(document, Exception):
             raise document
         return document, None

logprep/connector/file/input.py

@@ -2,7 +2,7 @@
 FileInput
 ==========
 A generic line input that returns the documents it was initialized with.
-If a "document" is derived from BaseException, that exception will be thrown instead of
+If a "document" is derived from Exception, that exception will be thrown instead of
 returning a document. The exception will be removed and subsequent calls may return documents or
 throw other exceptions in the given order.
 

logprep/connector/json/input.py

@@ -4,7 +4,7 @@
 
 A json input that returns the documents it was initialized with.
 
-If a "document" is derived from BaseException, that exception will be thrown instead of
+If a "document" is derived from Exception, that exception will be thrown instead of
 returning a document. The exception will be removed and subsequent calls may return documents or
 throw other exceptions in the given order.
 

logprep/connector/jsonl/input.py

@@ -4,7 +4,7 @@
 
 A json line input that returns the documents it was initialized with.
 
-If a "document" is derived from BaseException, that exception will be thrown instead of
+If a "document" is derived from Exception, that exception will be thrown instead of
 returning a document. The exception will be removed and subsequent calls may return documents or
 throw other exceptions in the given order.
 

logprep/filter/expression/filter_expression.py

@@ -6,7 +6,7 @@
 from typing import Any, List
 
 
-class FilterExpressionError(BaseException):
+class FilterExpressionError(Exception):
     """Base class for FilterExpression related exceptions."""
 
 

logprep/filter/lucene_filter.py

@@ -130,7 +130,7 @@
 logger = logging.getLogger("LuceneFilter")
 
 
-class LuceneFilterError(BaseException):
+class LuceneFilterError(Exception):
     """Base class for LuceneFilter related exceptions."""
 
 

logprep/processor/amides/detection.py

@@ -10,7 +10,7 @@
 from logprep.processor.amides.features import CommaSeparation
 
 
-class DetectionModelError(BaseException):
+class DetectionModelError(Exception):
     """Base exception class for all RuleModel-related errors."""
 
 
@@ -98,7 +98,7 @@ def detect(self, sample: str) -> Tuple[bool, float]:
         return False, round(confidence_value, 3)
 
 
-class RuleAttributorError(BaseException):
+class RuleAttributorError(Exception):
     """Base class for all RuleAttributor-related Errors."""
 
 

logprep/processor/dissector/processor.py

@@ -70,9 +70,7 @@ def _get_mappings(self, event, rule) -> List[Tuple[Callable, dict, str, str, str
                 if loop_content is None:
                     if rule.ignore_missing_fields:
                         continue
-                    error = BaseException(
-                        f"dissector: mapping field '{source_field}' does not exist"
-                    )
+                    error = Exception(f"dissector: mapping field '{source_field}' does not exist")
                     self._handle_warning_error(event, rule, error)
             if delimiter is not None and loop_content is not None:
                 content, _, loop_content = loop_content.partition(delimiter)

logprep/processor/field_manager/processor.py

@@ -179,8 +179,7 @@ def _get_field_values(event, source):
 
     def _get_missing_fields_error(self, source_fields, field_values):
         missing_fields = [key for key, value in zip(source_fields, field_values) if value is None]
-        error = BaseException(f"{self.name}: missing source_fields: {missing_fields}")
-        return error
+        return Exception(f"{self.name}: missing source_fields: {missing_fields}")
 
     @staticmethod
     def _get_flatten_source_fields(source_fields_values):

logprep/processor/labeler/labeling_schema.py

@@ -12,7 +12,7 @@
 from logprep.util.getter import GetterFactory
 
 
-class LabelingSchemaError(BaseException):
+class LabelingSchemaError(Exception):
     """Base class for LabelingSchema related exceptions."""
 
 

logprep/processor/pseudonymizer/processor.py

@@ -344,9 +344,9 @@ def _wrap_hash(self, hash_string: str) -> str:
     def _update_cache_metrics(self):
         cache_info_pseudonyms = self._get_pseudonym_dict_cached.cache_info()
         cache_info_urls = self._pseudonymize_url_cached.cache_info()
-        self.metrics.new_results = cache_info_pseudonyms.misses + cache_info_urls.misses
-        self.metrics.cached_results = cache_info_pseudonyms.hits + cache_info_urls.hits
-        self.metrics.num_cache_entries = cache_info_pseudonyms.currsize + cache_info_urls.currsize
-        self.metrics.cache_load = (cache_info_pseudonyms.currsize + cache_info_urls.currsize) / (
+        self.metrics.new_results += cache_info_pseudonyms.misses + cache_info_urls.misses
+        self.metrics.cached_results += cache_info_pseudonyms.hits + cache_info_urls.hits
+        self.metrics.num_cache_entries += cache_info_pseudonyms.currsize + cache_info_urls.currsize
+        self.metrics.cache_load += (cache_info_pseudonyms.currsize + cache_info_urls.currsize) / (
             cache_info_pseudonyms.maxsize + cache_info_urls.maxsize
         )

logprep/processor/template_replacer/processor.py

@@ -45,7 +45,7 @@
 from logprep.util.helper import add_field_to, get_dotted_field_value
 
 
-class TemplateReplacerError(BaseException):
+class TemplateReplacerError(Exception):
     """Base class for TemplateReplacer related exceptions."""
 
     def __init__(self, name: str, message: str):

logprep/util/auto_rule_tester/auto_rule_tester.py

@@ -77,7 +77,7 @@
 
 
 # pylint: disable=protected-access
-class AutoRuleTesterException(BaseException):
+class AutoRuleTesterException(Exception):
     """Base class for AutoRuleTester related exceptions."""
 
     def __init__(self, message: str):

logprep/util/grok_pattern_loader.py

@@ -6,7 +6,7 @@
 PATTERN_CONVERSION = [("[[:alnum:]]", r"\w")]
 
 
-class GrokPatternLoaderError(BaseException):
+class GrokPatternLoaderError(Exception):
     """Base class for GrokPatternLoader related exceptions."""
 
     def __init__(self, message: str):

logprep/util/pre_detector_rule_matching_tester.py

@@ -23,7 +23,7 @@
 
 
 # pylint: disable=protected-access
-class MatchingRuleTesterException(BaseException):
+class MatchingRuleTesterException(Exception):
     """Base class for MatchingRuleTester related exceptions."""
 
     def __init__(self, message: str):

tests/unit/connector/test_confluent_kafka_common.py

@@ -31,7 +31,7 @@ def test_create_fails_for_unknown_option(self):
     def test_error_callback_logs_error(self):
         self.object.metrics.number_of_errors = 0
         with mock.patch("logging.Logger.error") as mock_error:
-            test_error = BaseException("test error")
+            test_error = Exception("test error")
             self.object._error_callback(test_error)
             mock_error.assert_called()
             mock_error.assert_called_with(f"{self.object.describe()}: {test_error}")

tests/unit/connector/test_confluent_kafka_input.py

@@ -108,6 +108,15 @@ def test_batch_finished_callback_calls_store_offsets(self, _):
         kafka_consumer.store_offsets.assert_called()
         kafka_consumer.store_offsets.assert_called_with(message=message)
 
+    @mock.patch("logprep.connector.confluent_kafka.input.Consumer")
+    def test_batch_finished_callback_calls_store_offsets(self, _):
+        input_config = deepcopy(self.CONFIG)
+        kafka_input = Factory.create({"test": input_config})
+        kafka_consumer = kafka_input._consumer
+        kafka_input._last_valid_record = None
+        kafka_input.batch_finished_callback()
+        kafka_consumer.store_offsets.assert_not_called()
+
     @mock.patch("logprep.connector.confluent_kafka.input.Consumer")
     def test_batch_finished_callback_raises_input_warning_on_kafka_exception(self, _):
         input_config = deepcopy(self.CONFIG)
@@ -119,7 +128,7 @@ def raise_generator(return_sequence):
             return list(reversed(return_sequence)).pop()
 
         kafka_consumer.store_offsets.side_effect = raise_generator(return_sequence)
-        kafka_input._last_valid_records = {0: "message"}
+        kafka_input._last_valid_record = {0: "message"}
         with pytest.raises(InputWarning):
             kafka_input.batch_finished_callback()
 
@@ -225,7 +234,7 @@ def test_get_next_raises_critical_input_parsing_error(self):
 
     def test_commit_callback_raises_warning_error_and_counts_failures(self):
         with pytest.raises(InputWarning, match="Could not commit offsets"):
-            self.object._commit_callback(BaseException, ["topic_partition"])
+            self.object._commit_callback(Exception, ["topic_partition"])
             assert self.object._commit_failures == 1
 
     def test_commit_callback_counts_commit_success(self):

tests/unit/connector/test_dummy_input.py

@@ -10,7 +10,7 @@
 from tests.unit.connector.base import BaseInputTestCase
 
 
-class DummyError(BaseException):
+class DummyError(Exception):
     pass
 
 
@@ -44,9 +44,9 @@ def test_raises_exceptions_instead_of_returning_them_in_document(self):
 
     def test_raises_exceptions_instead_of_returning_them(self):
         config = copy.deepcopy(self.CONFIG)
-        config["documents"] = [BaseException]
+        config["documents"] = [Exception]
         self.object = Factory.create({"Test Instance Name": config})
-        with raises(BaseException):
+        with raises(Exception):
             self.object.get_next(self.timeout)
 
     def test_repeat_documents_repeats_documents(self):

tests/unit/connector/test_dummy_output.py

@@ -45,7 +45,7 @@ def test_raises_exception_on_call_to_store(self):
         config.update({"exceptions": ["FatalOutputError"]})
         dummy_output = Factory.create({"test connector": config})
 
-        with raises(BaseException, match="FatalOutputError"):
+        with raises(Exception, match="FatalOutputError"):
             dummy_output.store({"order": 0})
 
     def test_raises_exception_on_call_to_store_custom(self):

tests/unit/connector/test_json_input.py

@@ -11,7 +11,7 @@
 from tests.unit.connector.base import BaseInputTestCase
 
 
-class DummyError(BaseException):
+class DummyError(Exception):
     pass
 
 

tests/unit/processor/clusterer/test_clusterer_signature_phase.py

@@ -48,14 +48,14 @@ def test_apply_signature_engine(self):
     def test_exception_if_raw_text_with_start_tag():
         log_record = LogRecord(raw_text="Test log with start tag <+> must raise an exception")
         sign_engine = SignatureEngine()
-        with pytest.raises(BaseException, match=r"Start-tag <\+> in raw log message"):
+        with pytest.raises(Exception, match=r"Start-tag <\+> in raw log message"):
             sign_engine.run(log_record, LogSaltModeTestComposition.rules[0])
 
     @staticmethod
     def test_exception_if_raw_text_with_end_tag():
         log_record = LogRecord(raw_text="Test log with end tag </+> must raise an exception")
         sign_engine = SignatureEngine()
-        with pytest.raises(BaseException, match=r"End-tag </\+> in raw log message"):
+        with pytest.raises(Exception, match=r"End-tag </\+> in raw log message"):
             sign_engine.run(log_record, LogSaltModeTestComposition.rules[0])
 
     @staticmethod
@@ -64,7 +64,7 @@ def test_missing_end_tag_in_sig_text():
             "Test log with a start tag <+>, but a missing end tag, " "must raise an exception"
         )
         stp = SignatureTagParser()
-        with pytest.raises(BaseException):
+        with pytest.raises(Exception):
             stp.calculate_signature(sig_text)