limit same error log to 3 per minute

quickwit/quickwit-indexing/src/actors/doc_processor.rs

@@ -416,12 +416,13 @@ impl DocProcessor {
                     processed_docs.push(processed_doc);
                 }
                 Err(error) => {
-                    if self.rate_limiter.should_log() {
+                    let error_str = error.to_string();
+                    if self.rate_limiter.should_log(&error_str) {
                         warn!(
                             index_id = self.counters.index_id,
                             source_id = self.counters.source_id,
                             "{}",
-                            error
+                            error_str
                         );
                     }
                     self.counters.record_error(error, num_bytes as u64);

quickwit/quickwit-indexing/src/actors/log_rate_limiter.rs

@@ -16,39 +16,63 @@
 //
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
-//
 
+use std::collections::hash_map::DefaultHasher;
+use std::collections::HashMap;
+use std::hash::{Hash, Hasher};
 use std::time::{Duration, Instant};
 
 /// `RateLimiter` will allow `max_per_minute` calls per minute.
 ///
 /// It does not keep a sliding window, so it is not very precise, but very cheap.
+///
+/// It allows the same log at most 3 times per minute.
 pub struct LogRateLimiter {
     count: usize,
     start_time: Instant,
     max_per_minute: usize,
+    // we store only the hash of the error message to avoid storing the whole string.
+    errors_logged: HashMap<u64, usize>,
+}
+fn hash_string(input: &str) -> u64 {
+    let mut hasher = DefaultHasher::new();
+    input.hash(&mut hasher);
+    hasher.finish()
 }
-
 impl LogRateLimiter {
     pub fn new(max_per_minute: usize) -> Self {
         LogRateLimiter {
             count: 0,
             start_time: Instant::now(),
             max_per_minute,
+            errors_logged: HashMap::new(),
         }
     }
 
-    pub fn should_log(&mut self) -> bool {
+    pub fn should_log(&mut self, error: &str) -> bool {
+        let hash = hash_string(error);
         let now = Instant::now();
         let elapsed = now.duration_since(self.start_time);
 
         if elapsed > Duration::from_secs(60) {
             self.count = 0;
             self.start_time = now;
+            self.errors_logged.clear();
+            self.errors_logged.insert(hash, 1);
             true
         } else {
             self.count += 1;
-            self.count < self.max_per_minute
+            if self.count > self.max_per_minute {
+                return false;
+            }
+            let error_count = self.errors_logged.entry(hash).or_insert(0);
+            *error_count += 1;
+            // if we have logged the same error more than 3 times, we stop logging it in this
+            // minute.
+            if *error_count > 3 {
+                return false;
+            }
+            true
         }
     }
 }