feat: add dynamic url info

CHANGELOG.md

@@ -2,6 +2,7 @@
 ### Breaking
 
 ### Features
+* generic_adder: add support for templated http urls
 
 ### Improvements
 

logprep/abc/getter.py

@@ -114,10 +114,12 @@ def get_json(self) -> dict | list:
         content = self._resolve_content(raw)
         return self._parse_json(content)
 
-    def get_collection(self) -> dict | list:
+    def get_collection(self, content_field: str | None = None) -> dict | list:
         """Gets and parses the raw content to yaml or json"""
         content = self._resolve_content_by_content_type()
 
+        content = Getter._apply_content_field(content, content_field)
+
         if isinstance(content, str):
             content = self._parse_yaml_or_json(content)
 
@@ -143,18 +145,26 @@ def _parse_newline_separated_list(content: str) -> list:
         """Helper which tries to convert content to list"""
         return content.splitlines()
 
-    def get_list(self, content_field: str | None = None) -> list:
-        """Gets list and fails otherwise"""
-
-        content = self._resolve_content_by_content_type()
-
+    @staticmethod
+    def _apply_content_field(
+        content: dict | list | str, content_field: str | None = None
+    ) -> dict | list | str:
         if isinstance(content, dict) and content_field is not None:
             content = content[content_field]
         elif content_field is not None:
             raise ValueError(
                 f"Expected mapping type when content_field is set, got {type(content)}"
             )
 
+        return content
+
+    def get_list(self, content_field: str | None = None) -> list:
+        """Gets list and fails otherwise"""
+
+        content = self._resolve_content_by_content_type()
+
+        content = Getter._apply_content_field(content, content_field)
+
         if isinstance(content, str):
             content = self._parse_newline_separated_list(content)
 

logprep/ng/processor/generic_adder/processor.py

@@ -25,10 +25,13 @@
 """
 
 import typing
+from typing import Sequence
 
 from logprep.ng.abc.processor import Processor
+from logprep.processor.base.exceptions import ProcessingWarning
 from logprep.processor.base.rule import Rule
 from logprep.processor.generic_adder.rule import GenericAdderRule
+from logprep.util.getter import RefreshableGetter
 from logprep.util.helper import add_fields_to
 
 
@@ -37,8 +40,26 @@ class GenericAdder(Processor):
 
     rule_class = GenericAdderRule
 
+    @property
+    def _rules(self) -> Sequence[GenericAdderRule]:
+        """Returns all rules"""
+        return typing.cast(Sequence[GenericAdderRule], self.rules)
+
+    async def setup(self):
+        await super().setup()
+        for rule in self._rules:
+            rule.init_generic_adder(self._job_tag_for_cleanup)
+
     def _apply_rules(self, event: dict, rule: Rule) -> None:
         rule = typing.cast(GenericAdderRule, rule)
-        items_to_add = rule.add
+
+        try:
+            items_to_add = rule.add(event)
+        except Exception as error:
+            raise ProcessingWarning(str(error), rule, event) from error
         if items_to_add:
             add_fields_to(event, items_to_add, rule, rule.merge_with_target, rule.overwrite_target)
+
+    def _shut_down(self) -> None:
+        RefreshableGetter.remove_callbacks_for_tag(self._job_tag_for_cleanup)
+        return super()._shut_down()

logprep/processor/generic_adder/processor.py

@@ -25,10 +25,12 @@
 """
 
 import typing
+from typing import Sequence
 
 from logprep.abc.processor import Processor
-from logprep.processor.base.rule import Rule
+from logprep.processor.base.exceptions import ProcessingWarning
 from logprep.processor.generic_adder.rule import GenericAdderRule
+from logprep.util.getter import RefreshableGetter
 from logprep.util.helper import add_fields_to
 
 
@@ -37,8 +39,24 @@ class GenericAdder(Processor):
 
     rule_class = GenericAdderRule
 
-    def _apply_rules(self, event: dict, rule: Rule):
-        rule = typing.cast(GenericAdderRule, rule)
-        items_to_add = rule.add
+    @property
+    def _rules(self) -> Sequence[GenericAdderRule]:
+        """Returns all rules"""
+        return typing.cast(Sequence[GenericAdderRule], self.rules)
+
+    def setup(self):
+        super().setup()
+        for rule in self._rules:
+            rule.init_generic_adder(self._job_tag_for_cleanup)
+
+    def _apply_rules(self, event: dict, rule: GenericAdderRule):
+        try:
+            items_to_add = rule.add(event)
+        except Exception as error:
+            raise ProcessingWarning(str(error), rule, event) from error
         if items_to_add:
             add_fields_to(event, items_to_add, rule, rule.merge_with_target, rule.overwrite_target)
+
+    def _shut_down(self) -> None:
+        RefreshableGetter.remove_callbacks_for_tag(self._job_tag_for_cleanup)
+        return super()._shut_down()