fix: properly handle DecoderError to avoid pipeline failures (#932)

* add failing test to verify the bug

* treat DecoderError as warning; use proper warning handler

* refactor method to fix misleading parameter naming

* add CHANGELOG entry

* apply changes to ng

* fix outdated pydoc; improve CHANGELOG

* refine CHANGELOG

Author: mhoff

CHANGELOG.md

@@ -9,7 +9,8 @@
 * simplify Dockerfile and remove docker build support for `LOGPREP_VERSION`
 
 ### Bugfix
-* generic resolver now follows yaml standard and accepts a list instead of relying on the ordering of a dict
+* generic_resolver now follows yaml standard and accepts a list instead of relying on the ordering of a dict
+* decoder errors are handled properly as warnings instead of causing pipeline failures
 
 ## 18.0.1
 ### Breaking

logprep/ng/processor/decoder/processor.py

@@ -46,7 +46,7 @@
 from logprep.processor.decoder.decoders import DECODERS, DecoderError
 from logprep.processor.decoder.rule import DecoderRule
 from logprep.processor.field_manager.rule import FieldManagerRule
-from logprep.util.helper import FieldValue, add_fields_to
+from logprep.util.helper import FieldValue
 from logprep.util.typing import is_list_of
 
 
@@ -78,6 +78,5 @@ def _decode(
                 return [decoder(value) for value in source_field_values]
             raise DecoderError("can only decode string values")
         except DecoderError as error:
-            add_fields_to(event, {"tags": rule.failure_tags}, merge_with_target=True)
-            self._event.errors.append(error)
+            self._handle_warning_error(event, rule, error)
             return []

logprep/processor/decoder/processor.py

@@ -46,7 +46,7 @@
 from logprep.processor.decoder.rule import DecoderRule
 from logprep.processor.field_manager.processor import FieldManager
 from logprep.processor.field_manager.rule import FieldManagerRule
-from logprep.util.helper import FieldValue, add_fields_to
+from logprep.util.helper import FieldValue
 from logprep.util.typing import is_list_of
 
 
@@ -78,6 +78,5 @@ def _decode(
                 return [decoder(value) for value in source_field_values]
             raise DecoderError("can only decode string values")
         except DecoderError as error:
-            add_fields_to(event, {"tags": rule.failure_tags}, merge_with_target=True)
-            self.result.errors.append(error)
+            self._handle_warning_error(event, rule, error)
             return []

logprep/util/typing.py

@@ -8,19 +8,19 @@
 T = TypeVar("T")
 
 
-def is_list_of(val: list[Any], class_or_tuple: type[T]) -> TypeGuard[list[T]]:
+def is_list_of(val: list[Any], class_type: type[T]) -> TypeGuard[list[T]]:
     """Checks and statically asserts that a list has a specific element type
 
     Parameters
     ----------
     val : list[Any]
         The list to be checked
-    class_or_tuple : type[T]
+    class_type : type[T]
         The element type to be checked against
 
     Returns
     -------
     TypeGuard[list[T]]
         The type information that `val` is of this list type
     """
-    return all(isinstance(obj, class_or_tuple) for obj in val)
+    return all(isinstance(obj, class_type) for obj in val)

tests/unit/ng/processor/decoder/test_decoder.py

@@ -6,6 +6,8 @@
 import pytest
 
 from logprep.ng.event.log_event import LogEvent
+from logprep.processor.base.exceptions import ProcessingError, ProcessingWarning
+from logprep.util.typing import is_list_of
 from tests.unit.ng.processor.base import BaseProcessorTestCase
 
 
@@ -689,7 +691,13 @@ def test_testcases_failure_handling(self, rule, event, expected):
         self._load_rule(rule)
         event = LogEvent(event, original=b"")
         result = self.object.process(event)
-        assert result.errors or result.warnings
+        assert len(result.errors) > 0 or len(result.warnings) > 0
+        assert is_list_of(
+            result.errors, ProcessingError
+        ), f"ProcessingError expected: {result.errors}"
+        assert is_list_of(
+            result.warnings, ProcessingWarning
+        ), f"ProcessingWarning expected: {result.warnings}"
         assert event.data == expected
 
     def test_decodes_different_source_json_escaping(self):

tests/unit/processor/decoder/test_decoder.py

@@ -5,6 +5,8 @@
 
 import pytest
 
+from logprep.processor.base.exceptions import ProcessingError, ProcessingWarning
+from logprep.util.typing import is_list_of
 from tests.unit.processor.base import BaseProcessorTestCase
 
 
@@ -686,7 +688,13 @@ def test_testcases(self, rule, event, expected):
     def test_testcases_failure_handling(self, rule, event, expected):
         self._load_rule(rule)
         result = self.object.process(event)
-        assert result.errors or result.warnings
+        assert len(result.errors) > 0 or len(result.warnings) > 0
+        assert is_list_of(
+            result.errors, ProcessingError
+        ), f"ProcessingError expected: {result.errors}"
+        assert is_list_of(
+            result.warnings, ProcessingWarning
+        ), f"ProcessingWarning expected: {result.warnings}"
         assert event == expected
 
     def test_decodes_different_source_json_escaping(self):