[SLES-3059] feat(logs): keep HTTP library log lines out of Datadog

bottlecap/Cargo.toml

@@ -39,6 +39,7 @@ tokio = { version = "1.47", default-features = false, features = ["macros", "rt-
 tokio-util = { version = "0.7", default-features = false }
 tracing = { version = "0.1", default-features = false }
 tracing-core = { version = "0.1", default-features = false }
+tracing-log = { version = "0.2", default-features = false, features = ["std"] }
 tracing-subscriber = { version = "0.3", default-features = false, features = ["std", "registry", "fmt", "env-filter", "tracing-log"] }
 hmac = { version = "0.12", default-features = false }
 sha2 = { version = "0.10", default-features = false }

bottlecap/src/bin/bottlecap/main.rs

@@ -108,6 +108,7 @@ use tokio::time::Instant;
 use tokio::{sync::Mutex as TokioMutex, sync::mpsc::Sender};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, warn};
+use tracing_subscriber::util::SubscriberInitExt;
 use ustr::Ustr;
 
 fn resolve_dsm_service(
@@ -234,9 +235,14 @@ fn enable_logging_subsystem() {
     )
     .unwrap_or(LogLevel::Info);
 
+    let http_libraries_off: Vec<String> = logger::HTTP_LIBRARY_TARGETS
+        .iter()
+        .map(|target| format!("{target}=off"))
+        .collect();
     let (env_filter, invalid_entries) = build_env_filter(
         &format!(
-            "h2=off,hyper=off,reqwest=off,rustls=off,datadog-trace-mini-agent=off,{log_level:?}"
+            "{},datadog-trace-mini-agent=off,{log_level:?}",
+            http_libraries_off.join(",")
         ),
         &std::env::var(LEVEL_BY_TARGET_ENV_VAR).unwrap_or_default(),
     )
@@ -252,7 +258,8 @@ fn enable_logging_subsystem() {
         .without_time()
         .event_format(logger::Formatter)
         .finish();
-    tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");
+    // `init` also forwards records from crates that log with the `log` crate to this subscriber.
+    subscriber.init();
 
     if !invalid_entries.is_empty() {
         warn!("Ignoring invalid {LEVEL_BY_TARGET_ENV_VAR} entries: {invalid_entries:?}");

bottlecap/src/logger/README.md

@@ -14,3 +14,5 @@ A level is `off`, `error`, `warn`, `info`, `debug` or `trace`. For example, if `
 An entry matches each target that starts with its text. For an entry that is not valid, the extension logs a warning that names the entry, and skips it.
 
 Use the targets that `mod.rs` defines, such as `cold_start_duration`. Other targets are module paths and crate names, which change when the code moves.
+
+The logs of `h2`, `hyper`, `reqwest` and `rustls` appear only in CloudWatch Logs. The extension does not send them to Datadog, because these libraries log each request, including the requests that send the extension's logs. They are off by default. For example, `hyper_util=debug,h2=debug` shows each TCP connect and each HTTP/2 frame of the flushes to Datadog.

bottlecap/src/logger/mod.rs

@@ -1,5 +1,6 @@
-use std::fmt;
+use std::fmt::{self, Write as _};
 use tracing_core::{Event, Subscriber};
+use tracing_log::NormalizeEvent;
 use tracing_subscriber::fmt::{
     FmtContext, FormattedFields,
     format::{self, FormatEvent, FormatFields},
@@ -15,6 +16,42 @@ pub mod cold_start;
 /// Tracing target for the duration of each cold start step.
 pub const COLD_START_DURATION: &str = "cold_start_duration";
 
+/// HTTP libraries that log each request, including the requests that send the extension's own
+/// logs to Datadog. The extension does not send their lines to Datadog, so that a flush cannot
+/// log lines for the next flush to send.
+pub const HTTP_LIBRARY_TARGETS: &[&str] = &["h2", "hyper", "reqwest", "rustls"];
+
+/// Starts each line that the formatter writes for an [`HTTP_LIBRARY_TARGETS`] target.
+const HTTP_LIBRARY_LINE_PREFIX: &str = "{\"target\":\"";
+
+fn is_http_library_target(target: &str) -> bool {
+    HTTP_LIBRARY_TARGETS
+        .iter()
+        .any(|library| target.starts_with(library))
+}
+
+/// Returns true if `record`, an `extension` record from the Telemetry API, is a line that the
+/// formatter wrote for an [`HTTP_LIBRARY_TARGETS`] target. The message check keeps the lines of
+/// other extensions.
+#[must_use]
+pub fn is_http_library_record(record: &serde_json::Value) -> bool {
+    const MESSAGE_PREFIX: &str = "DD_EXTENSION | ";
+    match record {
+        serde_json::Value::String(line) => {
+            line.starts_with(HTTP_LIBRARY_LINE_PREFIX)
+                && line.contains(&format!(",\"message\":\"{MESSAGE_PREFIX}"))
+        }
+        serde_json::Value::Object(obj) => {
+            obj.contains_key("target")
+                && obj
+                    .get("message")
+                    .and_then(serde_json::Value::as_str)
+                    .is_some_and(|message| message.starts_with(MESSAGE_PREFIX))
+        }
+        _ => false,
+    }
+}
+
 /// Writes `s` to `w` with the 6 mandatory JSON string escape sequences applied.
 /// Handles: `"`, `\`, `\n`, `\r`, `\t`, and U+0000–U+001F control characters.
 fn write_json_escaped(w: &mut impl fmt::Write, s: &str) -> fmt::Result {
@@ -36,25 +73,40 @@ fn write_json_escaped(w: &mut impl fmt::Write, s: &str) -> fmt::Result {
 pub struct Formatter;
 
 /// Visitor that captures the message and the `f64` fields of a tracing event, so that Datadog
-/// parses those fields as numbers. It drops fields of other types. To write another type, add
-/// its `record_*` method.
+/// parses those fields as numbers. It keeps the other fields as `name=value` text, which only
+/// HTTP library lines write. To write another type, add its `record_*` method.
 #[derive(Default)]
 struct FieldVisitor {
     message: String,
     /// The `f64` fields, in the order of the event.
     fields: Vec<(&'static str, f64)>,
+    /// The other fields as ` name=value` text. h2, for example, logs each frame in a `frame` field.
+    other_fields: String,
+}
+
+impl FieldVisitor {
+    fn record_other(&mut self, field: &tracing::field::Field, value: fmt::Arguments<'_>) {
+        // `log.*` fields carry the target and location of a record from the `log` crate.
+        if !field.name().starts_with("log.") {
+            let _ = write!(self.other_fields, " {}={value}", field.name());
+        }
+    }
 }
 
 impl tracing::field::Visit for FieldVisitor {
     fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn fmt::Debug) {
         if field.name() == "message" {
             self.message = format!("{value:?}");
+        } else {
+            self.record_other(field, format_args!("{value:?}"));
         }
     }
 
     fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
         if field.name() == "message" {
             self.message = value.to_string();
+        } else {
+            self.record_other(field, format_args!("{value}"));
         }
     }
 
@@ -99,7 +151,9 @@ where
         mut writer: format::Writer<'_>,
         event: &Event<'_>,
     ) -> fmt::Result {
-        let metadata = event.metadata();
+        // A record from the `log` crate arrives with the target `log`, and its own target in a field.
+        let normalized = event.normalized_metadata();
+        let metadata = normalized.as_ref().unwrap_or_else(|| event.metadata());
         let level = metadata.level();
 
         let mut visitor = FieldVisitor::default();
@@ -122,12 +176,27 @@ where
             }
         }
 
-        let message = format!("DD_EXTENSION | {level} | {span_prefix}{}", visitor.message);
+        let is_http_library = is_http_library_target(metadata.target());
+        let other_fields = if is_http_library {
+            visitor.other_fields.as_str()
+        } else {
+            ""
+        };
+        let message = format!(
+            "DD_EXTENSION | {level} | {span_prefix}{}{other_fields}",
+            visitor.message
+        );
 
         // Setting specifically `status` as opposed to `level` since AWS
         // filters out logs by the `level` field. This allows our logs to
         // appear in CWL regardless of the log level.
-        write!(writer, "{{\"status\":\"{level}\",\"message\":\"")?;
+        writer.write_char('{')?;
+        if is_http_library {
+            writer.write_str("\"target\":\"")?;
+            write_json_escaped(&mut writer, metadata.target())?;
+            writer.write_str("\",")?;
+        }
+        write!(writer, "\"status\":\"{level}\",\"message\":\"")?;
         write_json_escaped(&mut writer, &message)?;
         writer.write_char('"')?;
         for (key, value) in fields_to_json(&visitor.fields) {
@@ -376,4 +445,76 @@ mod tests {
         assert!(parsed.get("count").is_none());
         assert!(parsed.get("name").is_none());
     }
+
+    #[test]
+    fn test_formatter_starts_http_library_line_with_target() {
+        let output = capture_log(|| {
+            tracing::debug!(
+                target: "hyper_util::client::legacy::connect::http",
+                "connecting to 34.149.115.168:443"
+            );
+        });
+
+        assert!(output.starts_with(
+            r#"{"target":"hyper_util::client::legacy::connect::http","status":"DEBUG","#
+        ));
+        let parsed: serde_json::Value = serde_json::from_str(output.trim()).unwrap();
+        assert!(is_http_library_record(&serde_json::Value::String(
+            output.trim().to_string()
+        )));
+        assert!(is_http_library_record(&parsed));
+    }
+
+    #[test]
+    fn test_formatter_uses_target_of_log_crate_record() {
+        let output = capture_log(|| {
+            tracing_log::format_trace(
+                &log::Record::builder()
+                    .target("rustls::client::hs")
+                    .level(log::Level::Debug)
+                    .args(format_args!("Not resuming any session"))
+                    .build(),
+            )
+            .unwrap();
+        });
+
+        assert!(output.starts_with(r#"{"target":"rustls::client::hs","status":"DEBUG","#));
+        assert!(output.contains(r#"DD_EXTENSION | DEBUG | Not resuming any session""#));
+    }
+
+    #[test]
+    fn test_formatter_writes_other_fields_only_for_http_library_lines() {
+        let output = capture_log(|| {
+            tracing::debug!(target: "h2::codec::framed_write", frame = ?("Headers", 1), "send");
+            tracing::debug!(target: "bottlecap::traces", attempt = 1_u64, "Attempting request");
+        });
+        let lines: Vec<&str> = output.lines().collect();
+
+        assert!(lines[0].contains(r#"DD_EXTENSION | DEBUG | send frame=(\"Headers\", 1)""#));
+        assert!(lines[1].contains(r#"DD_EXTENSION | DEBUG | Attempting request""#));
+    }
+
+    #[test]
+    fn test_formatter_writes_no_target_for_extension_line() {
+        let output = capture_log(|| {
+            tracing::debug!(target: "bottlecap::traces", "Flushing 1 traces");
+        });
+
+        assert!(output.starts_with(r#"{"status":"DEBUG","#));
+        assert!(!is_http_library_record(&serde_json::Value::String(
+            output.trim().to_string()
+        )));
+    }
+
+    #[test]
+    fn test_is_http_library_record_keeps_other_extension_lines() {
+        let other_extension = serde_json::json!({"target": "app", "message": "started"});
+        assert!(!is_http_library_record(&other_extension));
+        assert!(!is_http_library_record(&serde_json::Value::String(
+            r#"{"target":"app","message":"started"}"#.to_string()
+        )));
+        assert!(!is_http_library_record(&serde_json::Value::String(
+            "plain text line".to_string()
+        )));
+    }
 }

bottlecap/src/logs/lambda/processor.rs

@@ -12,6 +12,7 @@ use crate::event_bus::Event;
 use crate::extension::telemetry::events::ReportMetrics;
 use crate::extension::telemetry::events::{Status, TelemetryEvent, TelemetryRecord};
 use crate::lifecycle::invocation::context::Context as InvocationContext;
+use crate::logger;
 use crate::logs::aggregator_service::AggregatorHandle;
 use crate::logs::processor::{Processor, Rule};
 use crate::tags::provider;
@@ -242,6 +243,9 @@ impl LambdaProcessor {
                 Err("Unable to parse log".into())
             }
             TelemetryRecord::Extension(v) => {
+                if logger::is_http_library_record(&v) {
+                    return Err("HTTP library log, kept out of Datadog".into());
+                }
                 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),
@@ -2361,6 +2365,28 @@ mod tests {
         assert_eq!(intake_log.message.status, "debug");
     }
 
+    #[tokio::test]
+    async fn test_http_library_log_is_not_sent() {
+        let config = Arc::new(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, _rx) = tokio::sync::mpsc::channel(2);
+        let mut processor =
+            LambdaProcessor::new(tags_provider, Arc::clone(&config), tx.clone(), true);
+
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 47).unwrap(),
+            record: TelemetryRecord::Extension(Value::String(
+                r#"{"target":"hyper_util::client::legacy::connect::http","status":"DEBUG","message":"DD_EXTENSION | DEBUG | connecting to 34.149.115.168:443"}"#
+                    .to_string(),
+            )),
+        };
+        assert!(processor.make_log(event).await.is_err());
+    }
+
     #[tokio::test]
     async fn test_get_intake_log_non_string_level_defaults_to_info() {
         let config = Arc::new(config::Config {