Merge pull request #165 from Aiven-Open/aiqin-search-metric

Add a threshold for search if want to emit a metric

README.rst

@@ -279,6 +279,7 @@ Example configuration for a single reader::
           "pattern": "SENSITIVE",
           "replacement": "[REDACTED]"
         }],
+      "threshold_for_metric_emit": 10
       "tags": {
           "type": "container"
       }
@@ -376,7 +377,9 @@ Secret filters and searches can be made to use re2 as a regex engine by running
 Change this setting to true to emit metrics to the metrics host whenever a secret pattern is matched.
 This matching happens before other filtering to help catch secrets being leaked to disk.
 
-
+``threshold_for_metric_emit`` ( default: ``10``)
+For the regex searches in journalpump, if search takes longer than this value, default 10 seconds, a metric will be emitted.
+type: int unit: second
 
 Sender Configuration
 --------------------

journalpump/journalpump.py

@@ -164,6 +164,7 @@ def __init__(
         self.secret_filter_metrics = self._configure_secret_filter_metrics(config)
         self.secret_filter_metric_last_send = time.monotonic()
         self._is_ready = True
+        self.threshold_for_metric_emit = self._configure_threshold_for_metric_emit(config)
 
     def invalidate(self) -> None:
         """
@@ -557,6 +558,9 @@ def _validate_and_build_secret_filters(self, config):
 
         return secret_filters
 
+    def _configure_threshold_for_metric_emit(self, config):
+        return int(config.get("threshold_for_metric_emit", 10))
+
     def perform_searches(self, jobject):
         entry = jobject.entry
         results = {}
@@ -586,7 +590,16 @@ def perform_searches(self, jobject):
                             break
                         byte_fields[field] = line
 
+                start_time = time.perf_counter()
                 match = regex.search(line)
+                regex_search_duration = time.perf_counter() - start_time
+                if regex_search_duration > self.threshold_for_metric_emit:
+                    self.stats.gauge(
+                        metric="journal.perform_search_regex_duration",
+                        value=regex_search_duration,
+                        tags=self.make_tags({"regex": regex.pattern}),
+                    )
+                    self.log.info("Slow regex search: %s for duration %s seconds", regex, regex_search_duration)
                 if not match:
                     all_match = False
                     break

test/test_journalpump.py

@@ -320,6 +320,9 @@ def test_journal_reader_tagging(tmpdir):
     pump = JournalPump(journalpump_path)
     reader = pump.readers["system"]
 
+    # Mocking stats object
+    stats_mock = mock.MagicMock()
+
     # matching entry
     entry = JournalObject(
         entry={
@@ -329,7 +332,9 @@ def test_journal_reader_tagging(tmpdir):
             "SYSLOG_IDENTIFIER": "kernel",
         }
     )
-    result = reader.perform_searches(entry)
+    with mock.patch("time.perf_counter") as mock_perf_counter, mock.patch.object(reader, "stats", stats_mock):
+        mock_perf_counter.side_effect = [0, 11, 0, 1, 2, 3, 4, 5]
+        result = reader.perform_searches(entry)
     expected = {
         "kernel.cpu.temperature": {
             "cpu": "CPU0",
@@ -339,6 +344,9 @@ def test_journal_reader_tagging(tmpdir):
         }
     }
     assert result == expected
+    stats_mock.gauge.assert_called_once_with(
+        metric="journal.perform_search_regex_duration", value=11, tags={"reader": "system", "regex": "^(?P<from>.*)$"}
+    )
 
     # some fields are not matching
     entry = JournalObject(