fix(datadog_agent source): return 200 on empty object payload (#19093)

* fix(datadog_agent source): return 200 on empty object payload

* add test

src/sources/datadog_agent/logs.rs

@@ -67,8 +67,9 @@ pub(crate) fn decode_log_body(
     api_key: Option<Arc<str>>,
     source: &DatadogAgentSource,
 ) -> Result<Vec<Event>, ErrorMessage> {
-    if body.is_empty() {
+    if body.is_empty() || body.as_ref() == b"{}" {
         // The datadog agent may send an empty payload as a keep alive
+        // https://github.com/DataDog/datadog-agent/blob/5a6c5dd75a2233fbf954e38ddcc1484df4c21a35/pkg/logs/client/http/destination.go#L52
         debug!(
             message = "Empty payload ignored.",
             internal_log_rate_limit = true

src/sources/datadog_agent/tests.rs

@@ -120,6 +120,27 @@ fn test_decode_log_body() {
     QuickCheck::new().quickcheck(inner as fn(Vec<LogMsg>) -> TestResult);
 }
 
+#[test]
+fn test_decode_log_body_empty_object() {
+    let body = Bytes::from("{}");
+    let api_key = None;
+    let decoder = crate::codecs::Decoder::new(
+        Framer::Bytes(BytesDecoder::new()),
+        Deserializer::Bytes(BytesDeserializer),
+    );
+
+    let source = DatadogAgentSource::new(
+        true,
+        decoder,
+        "http",
+        test_logs_schema_definition(),
+        LogNamespace::Legacy,
+    );
+
+    let events = decode_log_body(body, api_key, &source).unwrap();
+    assert_eq!(events.len(), 0);
+}
+
 #[test]
 fn generate_config() {
     crate::test_util::test_generate_config::<DatadogAgentConfig>();