Record bytes fed to the scanner

sds/src/scanner/metrics.rs

@@ -49,6 +49,7 @@ impl RuleMetrics {
  * match_count: Number of matches found
  * suppressed_match_count: Number of matches suppressed
  * cpu_duration: Time spent in CPU operations
+ * scanned_bytes: Bytes fed to at least one rule
  *
  * In case of too high cardinality, please refer to https://github.com/DataDog/logs-backend/blob/prod/domains/commons/shared/libs/telemetry/src/main/java/com/dd/metrics/RegistryCacheTags.java
  */
@@ -57,6 +58,7 @@ pub struct ScannerMetrics {
     pub match_count: Counter,
     pub suppressed_match_count: Counter,
     pub cpu_duration: Counter,
+    pub scanned_bytes: Counter,
 }
 
 impl ScannerMetrics {
@@ -66,6 +68,7 @@ impl ScannerMetrics {
             match_count: counter!("scanning.match_count", labels.clone()),
             suppressed_match_count: counter!("scanning.suppressed_match_count", labels.clone()),
             cpu_duration: counter!("scanning.cpu_duration", labels.clone()),
+            scanned_bytes: counter!("scanning.scanned_bytes", labels.clone()),
         }
     }
 }

sds/src/scanner/mod.rs

@@ -1238,7 +1238,9 @@ impl<'a, E: Encoding> ContentVisitor<'a> for ScannerContentVisitor<'a, E> {
         let mut per_string_data = SharedData::new();
         let wildcard_indices_per_path = self.wildcarded_indexes.get(path);
 
-        rule_visitor.visit_rule_indices(|rule_index| {
+        // Count each string once. Rule fan-out is already scanning.cpu_duration.
+        let mut did_scan = false;
+        let visit_result = rule_visitor.visit_rule_indices(|rule_index| {
             if self.blocked_rules.contains(&rule_index) {
                 return Ok(());
             }
@@ -1284,6 +1286,7 @@ impl<'a, E: Encoding> ContentVisitor<'a> for ScannerContentVisitor<'a, E> {
                     scan_metadata: self.scan_metadata,
                 };
 
+                did_scan = true;
                 let async_status = rule.get_string_matches(content, path, &mut ctx)?;
 
                 match async_status {
@@ -1299,7 +1302,15 @@ impl<'a, E: Encoding> ContentVisitor<'a> for ScannerContentVisitor<'a, E> {
                 }
             }
             Ok(())
-        })?;
+        });
+
+        if did_scan {
+            self.scanner
+                .metrics
+                .scanned_bytes
+                .increment(content.len() as u64);
+        }
+        visit_result?;
 
         // If there are any matches, the string will need to be accessed to check for false positives from
         // excluded matches, any to potentially mutate the string.

sds/src/scanner/test/metrics.rs

@@ -1,7 +1,7 @@
 use crate::match_action::MatchAction;
 use crate::scanner::regex_rule::config::{ProximityKeywordsConfig, RegexRuleConfig};
 use crate::scanner::scope::Scope;
-use crate::scanner::{RootRuleConfig, ScannerBuilder};
+use crate::scanner::{RootRuleConfig, ScanOptions, ScannerBuilder};
 use crate::{Path, PathSegment, simple_event::SimpleEvent};
 use metrics::{Key, Label};
 use metrics_util::CompositeKey;
@@ -51,6 +51,12 @@ fn should_submit_scanning_metrics() {
         .expect("metric not found");
     assert_eq!(metric_value, &(None, None, DebugValue::Counter(1)));
 
+    let metric_name = "scanning.scanned_bytes";
+    let metric_value = snapshot
+        .get(&CompositeKey::new(Counter, Key::from_name(metric_name)))
+        .expect("metric not found");
+    assert_eq!(metric_value, &(None, None, DebugValue::Counter(13)));
+
     let metric_name = "scanning.cpu_duration";
     let metric_value = snapshot
         .get(&CompositeKey::new(Counter, Key::from_name(metric_name)))
@@ -61,6 +67,111 @@ fn should_submit_scanning_metrics() {
     }
 }
 
+#[test]
+fn should_count_scanned_bytes_once_when_multiple_rules_scan_the_same_string() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+
+    metrics::with_local_recorder(&recorder, || {
+        let rule_0 = RootRuleConfig::new(RegexRuleConfig::new("secret").build())
+            .match_action(MatchAction::None);
+        let rule_1 = RootRuleConfig::new(RegexRuleConfig::new("foo").build())
+            .match_action(MatchAction::None);
+
+        let scanner = ScannerBuilder::new(&[rule_0, rule_1]).build().unwrap();
+        let mut content = SimpleEvent::Map(BTreeMap::from([(
+            "key1".to_string(),
+            SimpleEvent::String("secret".to_string()),
+        )]));
+
+        scanner.scan(&mut content).unwrap();
+    });
+
+    let snapshot = snapshotter.snapshot().into_hashmap();
+
+    let metric_name = "scanning.scanned_bytes";
+    let metric_value = snapshot
+        .get(&CompositeKey::new(Counter, Key::from_name(metric_name)))
+        .expect("metric not found");
+    assert_eq!(metric_value, &(None, None, DebugValue::Counter(6)));
+}
+
+#[test]
+fn should_count_scanned_bytes_only_for_included_fields() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+
+    metrics::with_local_recorder(&recorder, || {
+        let rule_0 = RootRuleConfig::new(RegexRuleConfig::new("secret").build())
+            .scope(Scope::include(vec![Path::from(vec!["key1".into()])]))
+            .match_action(MatchAction::None);
+
+        let scanner = ScannerBuilder::new(&[rule_0]).build().unwrap();
+        let mut content = SimpleEvent::Map(BTreeMap::from([
+            (
+                "key1".to_string(),
+                SimpleEvent::String("secret".to_string()),
+            ),
+            (
+                "key2".to_string(),
+                SimpleEvent::String("secret-other".to_string()),
+            ),
+        ]));
+
+        scanner.scan(&mut content).unwrap();
+    });
+
+    let snapshot = snapshotter.snapshot().into_hashmap();
+
+    let metric_name = "scanning.scanned_bytes";
+    let metric_value = snapshot
+        .get(&CompositeKey::new(Counter, Key::from_name(metric_name)))
+        .expect("metric not found");
+    assert_eq!(metric_value, &(None, None, DebugValue::Counter(6)));
+}
+
+#[test]
+fn should_not_count_scanned_bytes_when_every_rule_is_blocked() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+
+    metrics::with_local_recorder(&recorder, || {
+        let rule_0 = RootRuleConfig::new(RegexRuleConfig::new("secret").build())
+            .match_action(MatchAction::None);
+
+        let scanner = ScannerBuilder::new(&[rule_0]).build().unwrap();
+        let mut content = SimpleEvent::Map(BTreeMap::from([(
+            "key1".to_string(),
+            SimpleEvent::String("secret".to_string()),
+        )]));
+
+        scanner
+            .scan_with_options(
+                &mut content,
+                ScanOptions {
+                    blocked_rules_idx: vec![0],
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+    });
+
+    let snapshot = snapshotter.snapshot().into_hashmap();
+
+    let metric_name = "scanned_events";
+    let metric_value = snapshot
+        .get(&CompositeKey::new(Counter, Key::from_name(metric_name)))
+        .expect("metric not found");
+    assert_eq!(metric_value, &(None, None, DebugValue::Counter(1)));
+
+    let metric_name = "scanning.scanned_bytes";
+    match snapshot.get(&CompositeKey::new(Counter, Key::from_name(metric_name))) {
+        None => {}
+        Some((_, _, DebugValue::Counter(0))) => {}
+        Some(other) => panic!("expected scanning.scanned_bytes to be absent or 0, got {other:?}"),
+    }
+}
+
 #[test]
 fn should_submit_match_count_metric_tagged_with_sds_rule_name() {
     let recorder = DebuggingRecorder::new();