feat(datadog_logs): optionally truncate oversized log events

Cargo.toml

@@ -223,6 +223,7 @@ serde_json = { version = "1.0.150", default-features = false, features = ["prese
 serde_path_to_error = "0.1.14"
 serde_with = { version = "3.21.0", default-features = false, features = ["std", "macros", "chrono_0_4"] }
 serde_yaml = { version = "0.9.34", default-features = false }
+simdutf8 = { version = "0.1.5", default-features = false }
 snafu = { version = "0.9.0", default-features = false, features = ["futures", "std"] }
 socket2 = { version = "0.6.3", default-features = false }
 tempfile = "3.27.0"

changelog.d/datadog_logs_truncate_oversized.enhancement.md

@@ -0,0 +1,7 @@
+The `datadog_logs` sink can now optionally truncate logs that exceed Datadog's
+per-log size limit. Configure `truncate_oversized_logs` to set the encoded log
+limit, mark shortened messages, and tag reduced logs. Logs whose non-message
+fields leave no room for a truncated message, or have no string message to
+truncate, are dropped.
+
+authors: bruceg

lib/vector-common/src/internal_event/metric_name.rs

@@ -109,6 +109,7 @@ pub enum CounterName {
     MemoryEnrichmentTableTtlExpirations,
     MemoryEnrichmentTableTtlExpirationsTotal,
     ComponentCpuUsageNsTotal,
+    DatadogLogsEventsTruncatedTotal,
     DatadogLogsReservedAttributeConflictsTotal,
     // Data-plane counter names emitted by the `host_metrics` source.
     CpuSecondsTotal,
@@ -503,6 +504,7 @@ impl CounterName {
                 "memory_enrichment_table_ttl_expirations_total"
             }
             Self::ComponentCpuUsageNsTotal => "component_cpu_usage_ns_total",
+            Self::DatadogLogsEventsTruncatedTotal => "datadog_logs_events_truncated_total",
             Self::DatadogLogsReservedAttributeConflictsTotal => {
                 "datadog_logs_reserved_attribute_conflicts_total"
             }

src/internal_events/datadog_logs.rs

@@ -4,6 +4,24 @@ use vector_lib::{
 };
 use vrl::path::OwnedTargetPath;
 
+#[derive(Debug, NamedInternalEvent)]
+pub struct DatadogLogsEventTruncated {
+    pub max_log_bytes: usize,
+    pub original_encoded_size: usize,
+}
+
+impl InternalEvent for DatadogLogsEventTruncated {
+    fn emit(self) {
+        debug!(
+            message = "Truncated a Datadog log event that exceeded the per-log size limit.",
+            max_log_bytes = self.max_log_bytes,
+            original_encoded_size = self.original_encoded_size,
+            internal_log_rate_limit = true,
+        );
+        counter!(CounterName::DatadogLogsEventsTruncatedTotal).increment(1);
+    }
+}
+
 #[derive(Debug, NamedInternalEvent)]
 pub struct DatadogLogsReservedAttributeConflict<'a> {
     pub meaning: &'static str,

src/sinks/datadog/logs/config.rs

@@ -34,6 +34,7 @@ use crate::{
 // of escaped double-quotes -- but we believe this should be very rare in
 // practice.
 pub const MAX_PAYLOAD_BYTES: usize = 5_000_000;
+pub(super) const DEFAULT_MAX_LOG_BYTES: usize = 1_000_000;
 pub const BATCH_HEADROOM_BYTES: usize = 750_000;
 pub const BATCH_MAX_EVENTS: usize = 1_000;
 pub const BATCH_DEFAULT_TIMEOUT_SECS: f64 = 5.0;
@@ -48,6 +49,18 @@ impl SinkBatchSettings for DatadogLogsDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = BATCH_DEFAULT_TIMEOUT_SECS;
 }
 
+/// Options for truncating logs that exceed Datadog's per-log size limit.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Derivative)]
+#[derivative(Default)]
+#[serde(deny_unknown_fields)]
+pub struct DatadogLogsTruncationConfig {
+    /// Maximum encoded size, in bytes, of a log before truncation is applied.
+    #[derivative(Default(value = "default_max_log_bytes()"))]
+    #[serde(default = "default_max_log_bytes")]
+    pub max_log_bytes: usize,
+}
+
 /// Configuration for the `datadog_logs` sink.
 #[configurable_component(sink("datadog_logs", "Publish log events to Datadog."))]
 #[derive(Clone, Debug, Derivative)]
@@ -81,17 +94,30 @@ pub struct DatadogLogsConfig {
     /// to not set it above 5,000,000 (5 MB, the standard Datadog API limit). Increase
     /// this when targeting a compatible endpoint that accepts larger payloads. The batch
     /// goal is derived as `max_payload_bytes - 750,000` bytes; events larger than the
-    /// batch goal are sent alone in their batch. Events exceeding `max_payload_bytes` are
-    /// dropped.
+    /// batch goal are sent alone in their batch. Single events that still exceed
+    /// `max_payload_bytes` after optional truncation are dropped.
     #[derivative(Default(value = "default_max_payload_bytes()"))]
     #[serde(default = "default_max_payload_bytes")]
     pub max_payload_bytes: Option<usize>,
+
+    /// Attempt to truncate logs whose encoded JSON exceeds `max_log_bytes`.
+    ///
+    /// The message is shortened to the largest size that fits and `...TRUNCATED...` is appended.
+    /// Every reduced log is tagged with `truncated:single_line`. Logs whose non-message fields
+    /// leave no room for a truncated message are sent unchanged if they fit
+    /// `max_payload_bytes`; the Datadog intake can further truncate them. Logs without a string
+    /// message to truncate are dropped.
+    pub truncate_oversized_logs: Option<DatadogLogsTruncationConfig>,
 }
 
 const fn default_max_payload_bytes() -> Option<usize> {
     Some(MAX_PAYLOAD_BYTES)
 }
 
+const fn default_max_log_bytes() -> usize {
+    DEFAULT_MAX_LOG_BYTES
+}
+
 const fn default_compression() -> Option<Compression> {
     Some(Compression::zstd_default())
 }
@@ -175,6 +201,7 @@ impl DatadogLogsConfig {
             self.max_payload_bytes.unwrap_or(MAX_PAYLOAD_BYTES),
         )
         .compression(self.compression.or_else(default_compression).unwrap())
+        .truncation(self.truncate_oversized_logs)
         .build();
 
         Ok(VectorSink::from_event_streamsink(sink))
@@ -253,6 +280,19 @@ impl ValidatedSink for DatadogLogsConfig {
             .into());
         }
 
+        if let Some(truncation) = self.truncate_oversized_logs {
+            let max_payload_bytes = self.max_payload_bytes.unwrap_or(MAX_PAYLOAD_BYTES);
+            // A single log is wrapped in `[` and `]`, so leave two bytes for the JSON array.
+            let maximum_log_bytes = max_payload_bytes - 2;
+            if !(1..=maximum_log_bytes).contains(&truncation.max_log_bytes) {
+                return Err(format!(
+                    "truncate_oversized_logs.max_log_bytes ({}) must be between 1 and {}",
+                    truncation.max_log_bytes, maximum_log_bytes
+                )
+                .into());
+            }
+        }
+
         let batch_goal_bytes =
             self.max_payload_bytes.unwrap_or(MAX_PAYLOAD_BYTES) - BATCH_HEADROOM_BYTES;
 
@@ -309,6 +349,46 @@ mod test {
         crate::test_util::test_generate_config::<DatadogLogsConfig>();
     }
 
+    #[test]
+    fn truncate_oversized_logs_accepts_empty_configuration() {
+        let config = serde_yaml::from_str::<DatadogLogsConfig>(indoc::indoc! {r#"
+            default_api_key: "test_key"
+            truncate_oversized_logs: {}
+        "#})
+        .expect("truncation configuration should deserialize");
+
+        let truncation = config
+            .truncate_oversized_logs
+            .expect("truncation should be enabled");
+        assert_eq!(truncation.max_log_bytes, 1_000_000);
+    }
+
+    #[test]
+    fn validate_rejects_truncation_limit_without_array_wrapper_room() {
+        let config: DatadogLogsConfig = serde_yaml::from_str(indoc::indoc! {r#"
+            default_api_key: "test_key"
+            max_payload_bytes: 800000
+            truncate_oversized_logs:
+              max_log_bytes: 799999
+        "#})
+        .unwrap();
+
+        assert!(config.validate().is_err());
+    }
+
+    #[test]
+    fn validate_accepts_custom_log_limit_above_datadog_default() {
+        let config: DatadogLogsConfig = serde_yaml::from_str(indoc::indoc! {r#"
+            default_api_key: "test_key"
+            max_payload_bytes: 3000000
+            truncate_oversized_logs:
+              max_log_bytes: 2000000
+        "#})
+        .unwrap();
+
+        assert!(config.validate().is_ok());
+    }
+
     #[test]
     fn validate_produces_usable_batch_settings() {
         let config = DatadogLogsConfig::default();

src/sinks/datadog/logs/mod.rs

@@ -5,10 +5,13 @@
 //! Datadog Log API. The log API is relatively generous in terms of its
 //! constraints, except that:
 //!
+//!   * an individual log over 1 MB is accepted but truncated
 //!   * a 'payload' is comprised of no more than 1,000 array members
 //!   * a 'payload' may not be more than 5Mb in size, uncompressed and
 //!   * a 'payload' may not mix API keys
 //!
+//! The sink can optionally perform per-log truncation before sending.
+//!
 //! Otherwise per [the
 //! docs](https://docs.datadoghq.com/api/latest/logs/#send-logs) there aren't
 //! other major constraints we have to follow in this implementation. The sink