fix config refresh interval (#818)

* set config refresh interval if no config change
* refactor filter argument for as_dict to `_unserializable_fields` class attribute
* add documentation for schedule_config_refresh
* update changelog
---------

Co-authored-by: Jörg Zimmermann <joerg.zimmermann@bwi.de>

Author: ekneg54

CHANGELOG.md

@@ -14,20 +14,26 @@
 ### Improvements
 
 * ensured that "_test.json" files are not loaded as rules
+* introduce new logger `Config`
+* refactor config refresh behavior from `logprep.runner` to `logprep.util.configuration`
+* refactor config related metrics from `logprep.runner` to `logprep.util.configuration`
+* added a log message for recovering config refresh mechanic from failing source
+
+### Bugfix
+
+* Fixed logging error in _revoke_callback() by adding error handling
+* Fixed endless loading in logprep test config
+* prevent the auto rule tester from loading rules directly defined inside the config, since they break the auto rule tester and can't have tests anyways
+* Fixed typo and broken link in documentation
+* Fixed assign_callback error in confluentkafka input
+* Fixed error logging in ` _get_configuration`, which caused the github checks to fail
+* Resolved `mypy` errors in `BaseProcessorTestCase.` by ensuring `self.object` and `self.patchers` are not `None` before accessing attributes.
+* Fix domain resolver errors for invalid domains
+* Fixed deprecation warnings caused by datetime when using Python >= 3.12
+* Fixed timestamp and timezone mismatch issue
+* Fixed a bug where config refresh interval was not reset to original interval after recovering from source related failures (i.e. http timeouts)
+* Fixed inconsistent generator statistics report during multithreading by making it thread safe
 
-### Bugfix
-
-- Fixed logging error in _revoke_callback() by adding error handling
-- Fixed endless loading in logprep test config
-- prevent the auto rule tester from loading rules directly defined inside the config, since they break the auto rule tester and can't have tests anyways
-- Fixed typo and broken link in documentation
-- Fixed assign_callback error in confluentkafka input
-- Fixed error logging in ` _get_configuration`, which caused the github checks to fail
-- Resolved `mypy` errors in `BaseProcessorTestCase.` by ensuring `self.object` and `self.patchers` are not `None` before accessing attributes.
-- Fix domain resolver errors for invalid domains
-- Fixed deprecation warnings caused by datetime when using Python >= 3.12
-- Fixed timestamp and timezone mismatch issue
-- Fixed inconsistent generator statistics report during multithreading by making it thread safe
 
 ## 16.1.0
 ### Deprecations

examples/exampledata/config/pipeline.yml

@@ -1,4 +1,4 @@
-version: 1
+version: 2
 process_count: 2
 timeout: 0.1
 restart_count: 2

logprep/runner.py

@@ -9,17 +9,11 @@
 from typing import Generator
 
 from attrs import define, field
-from schedule import Scheduler
 
 from logprep.abc.component import Component
 from logprep.framework.pipeline_manager import PipelineManager
-from logprep.metrics.metrics import CounterMetric, GaugeMetric
-from logprep.util.configuration import (
-    ConfigGetterException,
-    Configuration,
-    ConfigVersionDidNotChangeError,
-    InvalidConfigurationError,
-)
+from logprep.metrics.metrics import GaugeMetric
+from logprep.util.configuration import Configuration
 from logprep.util.defaults import EXITCODES
 
 
@@ -52,8 +46,6 @@ class Runner:
 
     _exit_received: bool = False
 
-    scheduler: Scheduler
-
     @define(kw_only=True)
     class Metrics(Component.Metrics):
         """Metrics for the Logprep Runner."""
@@ -66,44 +58,6 @@ class Metrics(Component.Metrics):
         )
         """Current size of the error queue."""
 
-        version_info: GaugeMetric = field(
-            factory=lambda: GaugeMetric(
-                description="Logprep version information",
-                name="version_info",
-                labels={"logprep": "unset", "config": "unset"},
-                inject_label_values=False,
-            )
-        )
-        """Logprep version info."""
-        config_refresh_interval: GaugeMetric = field(
-            factory=lambda: GaugeMetric(
-                description="Logprep config refresh interval",
-                name="config_refresh_interval",
-                labels={"from": "unset", "config": "unset"},
-            )
-        )
-        """Indicates the configuration refresh interval in seconds."""
-        number_of_config_refreshes: CounterMetric = field(
-            factory=lambda: CounterMetric(
-                description="Indicates how often the logprep configuration was updated.",
-                name="number_of_config_refreshes",
-                labels={"from": "unset", "config": "unset"},
-            )
-        )
-        """Indicates how often the logprep configuration was updated."""
-        number_of_config_refresh_failures: CounterMetric = field(
-            factory=lambda: CounterMetric(
-                description=(
-                    "Indicates how often the logprep configuration "
-                    "could not be updated due to failures during the update."
-                ),
-                name="number_of_config_refreshes",
-                labels={"from": "unset", "config": "unset"},
-            )
-        )
-        """Indicates how often the logprep configuration could not be updated
-          due to failures during the update."""
-
     @property
     def _metric_labels(self) -> dict[str, str]:
         labels = {
@@ -112,21 +66,6 @@ def _metric_labels(self) -> dict[str, str]:
         }
         return labels
 
-    @property
-    def _config_refresh_interval(self) -> int:
-        """Indicates the configuration refresh interval in seconds."""
-        return self._configuration.config_refresh_interval
-
-    @_config_refresh_interval.setter
-    def _config_refresh_interval(self, value: int | None) -> None:
-        """Set the configuration refresh interval in seconds."""
-        if value is None:
-            self._configuration.config_refresh_interval = None
-        elif value <= 5:
-            self._configuration.config_refresh_interval = 5
-        else:
-            self._configuration.config_refresh_interval = value
-
     # Use this method to obtain a runner singleton for production
     @staticmethod
     def get_runner(configuration: Configuration) -> "Runner":
@@ -137,89 +76,53 @@ def get_runner(configuration: Configuration) -> "Runner":
 
     # For production, use the get_runner method to create/get access to a singleton!
     def __init__(self, configuration: Configuration) -> None:
-        self._manager: PipelineManager | None = None
         atexit.register(self.stop_and_exit)
         self.exit_code = EXITCODES.SUCCESS
         self._configuration = configuration
+        self._config_version = self._configuration.version  # to trigger reloads
         self.metrics = self.Metrics(labels={"logprep": "unset", "config": "unset"})
         self._logger = logging.getLogger("Runner")
         self._manager = PipelineManager(configuration)
-        self.scheduler = Scheduler()
 
-    def start(self):
+    def start(self) -> None:
         """Start processing.
 
         This runs until an SIGTERM, SIGINT or KeyboardInterrupt signal is received, or an unhandled
         error occurs.
         """
-        self._set_version_info_metric()
-        self._schedule_config_refresh_job()
+        self._configuration.schedule_config_refresh()
         self._manager.start()
         self._logger.info("Startup complete")
         self._logger.debug("Runner iterating")
         self._iterate()
 
-    def stop_and_exit(self):
+    def stop_and_exit(self) -> None:
         """Stop the runner and exit the process."""
         self._logger.info("Shutting down")
         if self._manager:
             self._manager.stop()
 
-    def _iterate(self):
+    def _iterate(self) -> None:
         for _ in self._keep_iterating():
             if self._exit_received:
                 break
-            self.scheduler.run_pending()
+            self._configuration.refresh()
+            if self._configuration.version != self._config_version:
+                self._manager.reload()
+                self._config_version = self._configuration.version
             if self._manager.should_exit():
-                self.exit_code = EXITCODES.PIPELINE_ERROR.value
+                self.exit_code = EXITCODES.PIPELINE_ERROR
                 self._logger.error("Restart count exceeded. Exiting.")
                 sys.exit(self.exit_code)
             if self._manager.error_queue is not None:
                 self.metrics.number_of_events_in_error_queue += self._manager.error_queue.qsize()
             self._manager.restart_failed_pipeline()
 
-    def reload_configuration(self):
-        """Reloads the configuration"""
-        try:
-            self._configuration.reload()
-            self._logger.info("Successfully reloaded configuration")
-            self.metrics.number_of_config_refreshes += 1
-            self._manager.reload()
-            self._schedule_config_refresh_job()
-            self._logger.info(f"Configuration version: {self._configuration.version}")
-            self._set_version_info_metric()
-        except ConfigGetterException as error:
-            self._logger.warning(f"Failed to load configuration: {error}")
-            self.metrics.number_of_config_refresh_failures += 1
-            self._config_refresh_interval = int(self._config_refresh_interval / 4)
-            self._schedule_config_refresh_job()
-        except ConfigVersionDidNotChangeError as error:
-            self._logger.info(str(error))
-        except InvalidConfigurationError as error:
-            self._logger.error(str(error))
-            self.metrics.number_of_config_refresh_failures += 1
-
-    def _set_version_info_metric(self):
-        self.metrics.version_info.add_with_labels(
-            1,
-            {"logprep": f"{version('logprep')}", "config": self._configuration.version},
-        )
-
-    def stop(self):
+    def stop(self) -> None:
         """Stop the logprep runner. Is called by the signal handler
         in run_logprep.py."""
         self._exit_received = True
 
-    def _schedule_config_refresh_job(self):
-        refresh_interval = self._config_refresh_interval
-        scheduler = self.scheduler
-        if scheduler.jobs:
-            scheduler.cancel_job(scheduler.jobs[0])
-        if isinstance(refresh_interval, (float, int)):
-            self.metrics.config_refresh_interval += refresh_interval
-            scheduler.every(refresh_interval).seconds.do(self.reload_configuration)
-            self._logger.info(f"Config refresh interval is set to: {refresh_interval} seconds")
-
     def _keep_iterating(self) -> Generator:
         """Indicates whether the runner should keep iterating."""