fix: properly handle falsy resolved values in generic_resolver (#933)

* add tests for verifying the bug

* fix handling of falsy values and improve static typing

* fix mypy issues

* attempt to circumvent false mypy issue

* restrict usage of None

* support None values properly

* fix typing issues and adapt ng

* attempt to fix mypy issue

* fix typo in docs

* add CHANGELOG entries

* add tests for resolve_file and add more complex tests

* fix tests

* use MISSING to avoid double dict query

* update CHANGELOG to reflect the bugfix for resolve_from_file

* fix review findings

* replace isinstance with a more readable alternative

Author: mhoff

CHANGELOG.md

@@ -4,13 +4,15 @@
 ### Features
 * add uv as dependency management, including uv.lock
 * allow configuration (and auto-creation) of service accounts in helm chart
+* generic_resolver now handles all FieldValue types (including None)
 
 ### Improvements
 * 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
 * decoder errors are handled properly as warnings instead of causing pipeline failures
+* generic_resolver now properly handles falsy values in resolve_list and resolve_from_file
 
 ## 18.0.1
 ### Breaking

logprep/ng/processor/generic_resolver/processor.py

@@ -23,17 +23,27 @@
 .. automodule:: logprep.processor.generic_resolver.rule
 """
 
+import typing
+from copy import deepcopy
 from functools import cached_property, lru_cache
-from typing import Callable, Optional
+from typing import Callable
 
 from attrs import define, field, validators
 
 from logprep.metrics.metrics import GaugeMetric
 from logprep.ng.abc.processor import Processor
 from logprep.ng.processor.field_manager.processor import FieldManager
 from logprep.processor.base.exceptions import FieldExistsWarning
+from logprep.processor.base.rule import Rule
 from logprep.processor.generic_resolver.rule import GenericResolverRule
-from logprep.util.helper import add_fields_to, get_dotted_field_value
+from logprep.util.helper import (
+    MISSING,
+    FieldValue,
+    Missing,
+    add_fields_to,
+    get_dotted_field_value,
+)
+from logprep.util.typing import is_lru_cached
 
 
 class GenericResolver(FieldManager):
@@ -43,9 +53,7 @@ class GenericResolver(FieldManager):
     class Config(Processor.Config):
         """GenericResolver config"""
 
-        max_cache_entries: Optional[int] = field(
-            validator=validators.optional(validators.instance_of(int)), default=0
-        )
+        max_cache_entries: int = field(validator=validators.instance_of(int), default=0)
         """(Optional) Size of cache for results when resolving from a list.
         The cache can be disabled by setting this option to :code:`0`.
 
@@ -56,9 +64,7 @@ class Config(Processor.Config):
            and OOM situations caused by the generic resolver cache.
 
         """
-        cache_metrics_interval: Optional[int] = field(
-            validator=validators.optional(validators.instance_of(int)), default=1
-        )
+        cache_metrics_interval: int = field(validator=validators.instance_of(int), default=1)
         """(Optional) Cache metrics won't be updated immediately.
         Instead updating is skipped for a number of events before it's next update.
         :code:`cache_metrics_interval` sets the number of events between updates (default: 1)."""
@@ -103,25 +109,33 @@ class Metrics(FieldManager.Metrics):
 
     rule_class = GenericResolverRule
 
+    @property
+    def config(self) -> Config:
+        """Returns the typed GenericResolver.Config"""
+        return typing.cast(GenericResolver.Config, self._config)
+
     @property
     def max_cache_entries(self) -> int:
         """Returns the configured number of max_cache_entries"""
-        return self._config.max_cache_entries
+        return self.config.max_cache_entries
 
     @property
     def cache_metrics_interval(self) -> int:
         """Returns the configured cache_metrics_interval"""
-        return self._config.cache_metrics_interval
+        return self.config.cache_metrics_interval
 
     @cached_property
-    def _get_lru_cached_value_from_list(self) -> Callable[[GenericResolverRule, str], str | None]:
-        """Returns lru cashed method to retrieve values from list if configured"""
+    def _get_lru_cached_value_from_list(
+        self,
+    ) -> Callable[[GenericResolverRule, str], FieldValue | Missing]:
+        """Returns lru cached method to retrieve values from list if configured"""
         if self.max_cache_entries <= 0:
             return self._resolve_value_from_list
         return lru_cache(maxsize=self.max_cache_entries)(self._resolve_value_from_list)
 
-    def _apply_rules(self, event: dict, rule: GenericResolverRule) -> None:
+    def _apply_rules(self, event: dict, rule: Rule) -> None:
         """Apply the given rule to the current event"""
+        rule = typing.cast(GenericResolverRule, rule)
         source_field_values = [
             get_dotted_field_value(event, source_field)
             for source_field in rule.field_mapping.keys()
@@ -130,25 +144,30 @@ def _apply_rules(self, event: dict, rule: GenericResolverRule) -> None:
         conflicting_fields = []
         for source_field, target_field in rule.field_mapping.items():
             source_field_value = str(get_dotted_field_value(event, source_field))
-            content = self._find_content_of_first_matching_pattern(rule, source_field_value)
-            if not content:
+            resolved_content = self._find_content_of_first_matching_pattern(
+                rule, source_field_value
+            )
+            if resolved_content is MISSING:
                 continue
             current_content = get_dotted_field_value(event, target_field)
-            if isinstance(current_content, list) and content in current_content:
+            if isinstance(current_content, list) and resolved_content in current_content:
                 continue
+            if isinstance(resolved_content, (list, dict)):
+                resolved_content = deepcopy(resolved_content)
             try:
                 add_fields_to(
                     event,
                     fields={
                         target_field: (
-                            [content]
+                            [resolved_content]
                             if rule.merge_with_target and current_content is None
-                            else content
+                            else resolved_content
                         )
                     },
                     rule=rule,
                     merge_with_target=rule.merge_with_target,
                     overwrite_target=rule.overwrite_target,
+                    skip_none=False,
                 )
             except FieldExistsWarning as error:
                 conflicting_fields.extend(error.skipped_fields)
@@ -160,28 +179,28 @@ def _apply_rules(self, event: dict, rule: GenericResolverRule) -> None:
 
     def _find_content_of_first_matching_pattern(
         self, rule: GenericResolverRule, source_field_value: str
-    ) -> str | None:
+    ) -> FieldValue | Missing:
         if rule.resolve_from_file:
             matches = rule.pattern.match(source_field_value)
             if matches:
                 mapping = matches.group("mapping")
                 if rule.ignore_case:
                     mapping = mapping.upper()
-                content = rule.additions.get(mapping)
-                if content:
+                content = rule.additions.get(mapping, MISSING)
+                if content is not MISSING:
                     return content
         return self._get_lru_cached_value_from_list(rule, source_field_value)
 
     def _resolve_value_from_list(
         self, rule: GenericResolverRule, source_field_value: str
-    ) -> Optional[str]:
+    ) -> FieldValue | Missing:
         for pattern, content in rule.compiled_resolve_list:
             if pattern.search(source_field_value):
                 return content
-        return None
+        return MISSING
 
     def _update_cache_metrics(self) -> None:
-        if self.max_cache_entries <= 0:
+        if not is_lru_cached(self._get_lru_cached_value_from_list):
             return
         self._cache_metrics_skip_count += 1
         if self._cache_metrics_skip_count < self.cache_metrics_interval:
@@ -192,7 +211,7 @@ def _update_cache_metrics(self) -> None:
         self.metrics.new_results += cache_info.misses
         self.metrics.cached_results += cache_info.hits
         self.metrics.num_cache_entries += cache_info.currsize
-        self.metrics.cache_load += cache_info.currsize / cache_info.maxsize
+        self.metrics.cache_load += cache_info.currsize / self.max_cache_entries
 
     def setup(self) -> None:
         super().setup()

logprep/processor/generic_resolver/processor.py

@@ -23,17 +23,27 @@
 .. automodule:: logprep.processor.generic_resolver.rule
 """
 
+import typing
+from copy import deepcopy
 from functools import cached_property, lru_cache
-from typing import Optional
+from typing import Callable
 
 from attrs import define, field, validators
 
 from logprep.abc.processor import Processor
 from logprep.metrics.metrics import GaugeMetric
 from logprep.processor.base.exceptions import FieldExistsWarning
+from logprep.processor.base.rule import Rule
 from logprep.processor.field_manager.processor import FieldManager
 from logprep.processor.generic_resolver.rule import GenericResolverRule
-from logprep.util.helper import add_fields_to, get_dotted_field_value
+from logprep.util.helper import (
+    MISSING,
+    FieldValue,
+    Missing,
+    add_fields_to,
+    get_dotted_field_value,
+)
+from logprep.util.typing import is_lru_cached
 
 
 class GenericResolver(FieldManager):
@@ -43,9 +53,7 @@ class GenericResolver(FieldManager):
     class Config(Processor.Config):
         """GenericResolver config"""
 
-        max_cache_entries: Optional[int] = field(
-            validator=validators.optional(validators.instance_of(int)), default=0
-        )
+        max_cache_entries: int = field(validator=validators.instance_of(int), default=0)
         """(Optional) Size of cache for results when resolving from a list.
         The cache can be disabled by setting this option to :code:`0`.
 
@@ -56,9 +64,7 @@ class Config(Processor.Config):
            and OOM situations caused by the generic resolver cache.
 
         """
-        cache_metrics_interval: Optional[int] = field(
-            validator=validators.optional(validators.instance_of(int)), default=1
-        )
+        cache_metrics_interval: int = field(validator=validators.instance_of(int), default=1)
         """(Optional) Cache metrics won't be updated immediately.
         Instead updating is skipped for a number of events before it's next update.
         :code:`cache_metrics_interval` sets the number of events between updates (default: 1)."""
@@ -104,24 +110,32 @@ class Metrics(FieldManager.Metrics):
     rule_class = GenericResolverRule
 
     @property
-    def max_cache_entries(self):
+    def config(self) -> Config:
+        """Returns the typed GenericResolver.Config"""
+        return typing.cast(GenericResolver.Config, self._config)
+
+    @property
+    def max_cache_entries(self) -> int:
         """Returns the configured number of max_cache_entries"""
-        return self._config.max_cache_entries
+        return self.config.max_cache_entries
 
     @property
-    def cache_metrics_interval(self):
+    def cache_metrics_interval(self) -> int:
         """Returns the configured cache_metrics_interval"""
-        return self._config.cache_metrics_interval
+        return self.config.cache_metrics_interval
 
     @cached_property
-    def _get_lru_cached_value_from_list(self):
-        """Returns lru cashed method to retrieve values from list if configured"""
+    def _get_lru_cached_value_from_list(
+        self,
+    ) -> Callable[[GenericResolverRule, str], FieldValue | Missing]:
+        """Returns lru cached method to retrieve values from list if configured"""
         if self.max_cache_entries <= 0:
             return self._resolve_value_from_list
         return lru_cache(maxsize=self.max_cache_entries)(self._resolve_value_from_list)
 
-    def _apply_rules(self, event: dict, rule: GenericResolverRule) -> None:
+    def _apply_rules(self, event: dict, rule: Rule) -> None:
         """Apply the given rule to the current event"""
+        rule = typing.cast(GenericResolverRule, rule)
         source_field_values = [
             get_dotted_field_value(event, source_field)
             for source_field in rule.field_mapping.keys()
@@ -130,25 +144,30 @@ def _apply_rules(self, event: dict, rule: GenericResolverRule) -> None:
         conflicting_fields = []
         for source_field, target_field in rule.field_mapping.items():
             source_field_value = str(get_dotted_field_value(event, source_field))
-            content = self._find_content_of_first_matching_pattern(rule, source_field_value)
-            if not content:
+            resolved_content = self._find_content_of_first_matching_pattern(
+                rule, source_field_value
+            )
+            if resolved_content is MISSING:
                 continue
             current_content = get_dotted_field_value(event, target_field)
-            if isinstance(current_content, list) and content in current_content:
+            if isinstance(current_content, list) and resolved_content in current_content:
                 continue
+            if isinstance(resolved_content, (list, dict)):
+                resolved_content = deepcopy(resolved_content)
             try:
                 add_fields_to(
                     event,
                     fields={
                         target_field: (
-                            [content]
+                            [resolved_content]
                             if rule.merge_with_target and current_content is None
-                            else content
+                            else resolved_content
                         )
                     },
                     rule=rule,
                     merge_with_target=rule.merge_with_target,
                     overwrite_target=rule.overwrite_target,
+                    skip_none=False,
                 )
             except FieldExistsWarning as error:
                 conflicting_fields.extend(error.skipped_fields)
@@ -160,28 +179,28 @@ def _apply_rules(self, event: dict, rule: GenericResolverRule) -> None:
 
     def _find_content_of_first_matching_pattern(
         self, rule: GenericResolverRule, source_field_value: str
-    ) -> str | None:
+    ) -> FieldValue | Missing:
         if rule.resolve_from_file:
             matches = rule.pattern.match(source_field_value)
             if matches:
                 mapping = matches.group("mapping")
                 if rule.ignore_case:
                     mapping = mapping.upper()
-                content = rule.additions.get(mapping)
-                if content:
+                content = rule.additions.get(mapping, MISSING)
+                if content is not MISSING:
                     return content
         return self._get_lru_cached_value_from_list(rule, source_field_value)
 
     def _resolve_value_from_list(
         self, rule: GenericResolverRule, source_field_value: str
-    ) -> Optional[str]:
+    ) -> FieldValue | Missing:
         for pattern, content in rule.compiled_resolve_list:
             if pattern.search(source_field_value):
                 return content
-        return None
+        return MISSING
 
-    def _update_cache_metrics(self):
-        if self.max_cache_entries <= 0:
+    def _update_cache_metrics(self) -> None:
+        if not is_lru_cached(self._get_lru_cached_value_from_list):
             return
         self._cache_metrics_skip_count += 1
         if self._cache_metrics_skip_count < self.cache_metrics_interval:
@@ -192,8 +211,8 @@ def _update_cache_metrics(self):
         self.metrics.new_results += cache_info.misses
         self.metrics.cached_results += cache_info.hits
         self.metrics.num_cache_entries += cache_info.currsize
-        self.metrics.cache_load += cache_info.currsize / cache_info.maxsize
+        self.metrics.cache_load += cache_info.currsize / self.max_cache_entries
 
-    def setup(self):
+    def setup(self) -> None:
         super().setup()
         self._cache_metrics_skip_count = 0