[SVLS-8231] fixing log requestId for LMI (#973)

https://datadoghq.atlassian.net/browse/SVLS-8231

## Overview
* Bug fix - When running a LMI function, function logs don't get
attributed with `lambda.requestId`. As a result, no function logs appear
for an invocation. They still show up for the function, just not not on
the invocation.
* We could have alternatively updated the UI to search for
`lambda.requestId` and `requestId` instead of just searching for
`lambda.requestId`. However, I would prefer for On Demand and Lambda
Managed Instance Functions to be consistent. So, both using
lambda.requestId is preferred.

## Testing 
* Added integration tests that checks for logs per invocation request.
* Also confirmed in the UI we are seeing the logs for an invocation
(before/after screenshot below).

<img width="1178" height="467" alt="Screenshot 2025-12-30 at 8 27 34 AM"
src="https://github.com/user-attachments/assets/3d9d8c50-6ede-4bae-bcab-a3cf3fed9f32"
/>

<img width="1178" height="467" alt="Screenshot 2025-12-30 at 8 20 34 AM"
src="https://github.com/user-attachments/assets/2e247e6a-bc9c-47c6-8f66-5cf2d57633c6"
/>

Author: jchrostek-dd

bottlecap/src/logs/lambda/processor.rs

@@ -89,7 +89,44 @@ impl LambdaProcessor {
     async fn get_message(&mut self, event: TelemetryEvent) -> Result<Message, Box<dyn Error>> {
         let copy = event.clone();
         match event.record {
-            TelemetryRecord::Function(v) | TelemetryRecord::Extension(v) => {
+            TelemetryRecord::Function(v) => {
+                let (request_id, message) = match v {
+                    serde_json::Value::Object(obj) => {
+                        let request_id = if self.is_managed_instance_mode {
+                            obj.get("requestId")
+                                .or_else(|| obj.get("AWSRequestId"))
+                                .and_then(|v| v.as_str())
+                                .map(ToString::to_string)
+                        } else {
+                            None
+                        };
+                        let msg = Some(serde_json::to_string(&obj).unwrap_or_default());
+                        (request_id, msg)
+                    },
+                    serde_json::Value::String(s) => (None, Some(s)),
+                    _ => (None, None),
+                };
+
+                if let Some(message) = message {
+                    if is_oom_error(&message) {
+                        debug!("LOGS | Got a runtime-specific OOM error. Incrementing OOM metric.");
+                        if let Err(e) = self.event_bus.send(Event::OutOfMemory(event.time.timestamp())).await {
+                            error!("LOGS | Failed to send OOM event to the main event bus: {e}");
+                        }
+                    }
+
+                    return Ok(Message::new(
+                        message,
+                        request_id,
+                        self.function_arn.clone(),
+                        event.time.timestamp_millis(),
+                        None,
+                    ));
+                }
+
+                Err("Unable to parse log".into())
+            }
+            TelemetryRecord::Extension(v) => {
                 let message = match v {
                     serde_json::Value::Object(obj) => Some(serde_json::to_string(&obj).unwrap_or_default()),
                     serde_json::Value::String(s) => Some(s),
@@ -1495,4 +1532,104 @@ mod tests {
         assert_eq!(log2.message.lambda.request_id, None);
         assert_eq!(processor.orphan_logs.len(), 0);
     }
+
+    #[tokio::test]
+    async fn test_lmi_extracts_request_id_from_function_json() {
+        let tags = HashMap::from([("test".to_string(), "tags".to_string())]);
+        let config = Arc::new(config::Config {
+            service: Some("test-service".to_string()),
+            tags: tags.clone(),
+            ..config::Config::default()
+        });
+
+        let tags_provider = Arc::new(provider::Provider::new(
+            Arc::clone(&config),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::from([("function_arn".to_string(), "test-arn".to_string())]),
+        ));
+
+        let (tx, _) = tokio::sync::mpsc::channel(2);
+
+        let mut processor = LambdaProcessor::new(
+            tags_provider,
+            Arc::new(config::Config {
+                service: Some("test-service".to_string()),
+                tags,
+                ..config::Config::default()
+            }),
+            tx.clone(),
+            true, // LMI mode
+        );
+
+        // Test with "requestId" field
+        let mut obj = serde_json::Map::new();
+        obj.insert(
+            "requestId".to_string(),
+            Value::String("test-request-123".to_string()),
+        );
+        obj.insert(
+            "message".to_string(),
+            Value::String("Hello World".to_string()),
+        );
+
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 47).unwrap(),
+            record: TelemetryRecord::Function(Value::Object(obj)),
+        };
+
+        let result = processor.get_message(event).await.unwrap();
+        assert_eq!(
+            result.lambda.request_id,
+            Some("test-request-123".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn test_regular_lambda_does_not_extract_request_id() {
+        let tags = HashMap::from([("test".to_string(), "tags".to_string())]);
+        let config = Arc::new(config::Config {
+            service: Some("test-service".to_string()),
+            tags: tags.clone(),
+            ..config::Config::default()
+        });
+
+        let tags_provider = Arc::new(provider::Provider::new(
+            Arc::clone(&config),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::from([("function_arn".to_string(), "test-arn".to_string())]),
+        ));
+
+        let (tx, _) = tokio::sync::mpsc::channel(2);
+
+        let mut processor = LambdaProcessor::new(
+            tags_provider,
+            Arc::new(config::Config {
+                service: Some("test-service".to_string()),
+                tags,
+                ..config::Config::default()
+            }),
+            tx.clone(),
+            false, // Regular Lambda mode (not LMI)
+        );
+
+        // Test that requestId is NOT extracted in regular Lambda mode
+        let mut obj = serde_json::Map::new();
+        obj.insert(
+            "requestId".to_string(),
+            Value::String("test-request-789".to_string()),
+        );
+        obj.insert(
+            "message".to_string(),
+            Value::String("Hello World".to_string()),
+        );
+
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 47).unwrap(),
+            record: TelemetryRecord::Function(Value::Object(obj)),
+        };
+
+        let result = processor.get_message(event).await.unwrap();
+        // Should be None because we're not in LMI mode
+        assert_eq!(result.lambda.request_id, None);
+    }
 }

integration-tests/tests/lmi.test.ts

@@ -34,8 +34,7 @@ describe('LMI Integration Tests', () => {
       expect(results[runtime].logs!.length).toBeGreaterThan(0);
     });
 
-    // SVLS-8231
-    test.failing('should have "Hello World!" log message', () => {
+    it('should have "Hello World!" log message', () => {
       expect(results[runtime].logs).toBeDefined();
       const helloWorldLog = results[runtime].logs!.find((log: any) =>
         log.message && log.message.includes('Hello World!')