Fix crash on TimeParserException during preprocessing caused by invalid timestamps (#862)

* Handle TimeParserException in get_next() to catch invalid time format values

Author: ppcad

CHANGELOG.md

@@ -45,6 +45,7 @@
 ### Bugfix
 
 * add `@timestamp` field to error documents
+* fix crash on `TimeParserException` during preprocessing caused by invalid timestamps
 
 ## 17.0.0
 ### Breaking

logprep/abc/input.py

@@ -21,7 +21,7 @@
 from logprep.metrics.metrics import Metric
 from logprep.processor.base.exceptions import FieldExistsWarning
 from logprep.util.helper import add_fields_to, get_dotted_field_value
-from logprep.util.time import UTC, TimeParser
+from logprep.util.time import UTC, TimeParser, TimeParserException
 from logprep.util.validators import dict_structure_validator
 
 
@@ -338,7 +338,7 @@ def get_next(self, timeout: float) -> dict | None:
                 self._add_arrival_timedelta_information_to_event(event)
             if self._add_env_enrichment:
                 self._add_env_enrichment_to_event(event)
-        except FieldExistsWarning as error:
+        except (FieldExistsWarning, TimeParserException) as error:
             raise CriticalInputError(self, error.args[0], event) from error
         return event
 

tests/unit/connector/base.py

@@ -582,6 +582,34 @@ def test_pipeline_preprocessing_does_not_add_timestamp_delta_if_configured_but_l
         result = connector.get_next(0.01)
         assert result == {"any": "content"}
 
+    @pytest.mark.parametrize(
+        ["timestamp", "expected_error_message", "message"],
+        [
+            ("0000-00-00 00:00:00", "out of range", "Timestamp is out of range"),
+            ("invalid", "Invalid isoformat", "Timestamp is invalid"),
+        ],
+    )
+    def test_pipeline_preprocessing_raises_criticalinputerror_on_timeparser_error(
+        self, timestamp, expected_error_message, message
+    ):
+        preprocessing_config = {
+            "preprocessing": {
+                "log_arrival_time_target_field": "arrival_time",
+                "log_arrival_timedelta": {
+                    "target_field": "log_arrival_timedelta",
+                    "reference_field": "@timestamp",
+                },
+            }
+        }
+        connector_config = deepcopy(self.CONFIG)
+        connector_config.update(preprocessing_config)
+        connector = Factory.create({"test connector": connector_config})
+        test_event = {"any": "content", "@timestamp": timestamp}
+        connector._get_event = mock.MagicMock(return_value=(test_event, None))
+        with pytest.raises(CriticalInputError, match=expected_error_message) as error:
+            _, _ = connector.get_next(0.01)
+        assert error.value.raw_input, message
+
     def test_preprocessing_enriches_by_env_variable(self):
         preprocessing_config = {
             "preprocessing": {