feat(bottlecap): handle OOM exceptions

bottlecap/src/bin/bottlecap/main.rs

@@ -400,7 +400,12 @@ async fn extension_loop_active(
                     match event {
                         Event::Metric(event) => {
                             debug!("Metric event: {:?}", event);
-                        }
+                        },
+                        Event::OutOfMemory => {
+                            let mut p = invocation_processor.lock().await;
+                            p.on_out_of_memory_error();
+                            drop(p);
+                        },
                         Event::Telemetry(event) =>
                             match event.record {
                                 TelemetryRecord::PlatformInitStart { .. } => {

bottlecap/src/events/mod.rs

@@ -19,4 +19,5 @@ impl MetricEvent {
 pub enum Event {
     Metric(MetricEvent),
     Telemetry(TelemetryEvent),
+    OutOfMemory,
 }

bottlecap/src/lifecycle/invocation/processor.rs

@@ -602,4 +602,8 @@ impl Processor {
             // todo: handle timeout
         }
     }
+
+    pub fn on_out_of_memory_error(&mut self) {
+        self.enhanced_metrics.increment_oom_metric();
+    }
 }

bottlecap/src/logs/lambda/processor.rs

@@ -33,6 +33,22 @@ pub struct LambdaProcessor {
     event_bus: Sender<Event>,
 }
 
+const OOM_ERRORS: [&str; 7] = [
+    "fatal error: runtime: out of memory",       // Go
+    "java.lang.OutOfMemoryError",                // Java
+    "JavaScript heap out of memory",             // Node
+    "Runtime exited with error: signal: killed", // Node
+    "MemoryError",                               // Python
+    "failed to allocate memory (NoMemoryError)", // Ruby
+    "OutOfMemoryException",                      // .NET
+];
+
+fn is_oom_error(error_msg: &str) -> bool {
+    OOM_ERRORS
+        .iter()
+        .any(|&oom_str| error_msg.contains(oom_str))
+}
+
 impl Processor<IntakeLog> for LambdaProcessor {}
 
 impl LambdaProcessor {
@@ -71,16 +87,22 @@ impl LambdaProcessor {
                     _ => None,
                 };
 
-                if message.is_none() {
-                    return Err("Unable to parse log".into());
+                if let Some(message) = message {
+                    if is_oom_error(&message) {
+                        if let Err(e) = self.event_bus.send(Event::OutOfMemory).await {
+                            error!("Failed to send OOM event to the main event bus: {e}");
+                        }
+                    }
+
+                    return Ok(Message::new(
+                        message,
+                        None,
+                        self.function_arn.clone(),
+                        event.time.timestamp_millis(),
+                    ));
                 }
 
-                Ok(Message::new(
-                    message.expect("infallible"),
-                    None,
-                    self.function_arn.clone(),
-                    event.time.timestamp_millis(),
-                ))
+                Err("Unable to parse log".into())
             }
             TelemetryRecord::PlatformInitStart {
                 runtime_version,

bottlecap/src/metrics/enhanced/lambda.rs

@@ -82,6 +82,10 @@ impl Lambda {
         self.increment_metric(constants::TIMEOUTS_METRIC);
     }
 
+    pub fn increment_oom_metric(&self) {
+        self.increment_metric(constants::OUT_OF_MEMORY_METRIC);
+    }
+
     pub fn set_init_duration_metric(&self, init_duration_ms: f64) {
         if !self.config.enhanced_metrics {
             return;