From 8d234483d2c025dec2bf033eaa533c80823aabd5 Mon Sep 17 00:00:00 2001
From: Loakesh Indiran <66478092+Loaki07@users.noreply.github.com>
Date: Wed, 18 Dec 2024 18:08:31 +0530
Subject: [PATCH] fix: check invalid stream_name during bulk ingestion (#5403)

#5378

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **Bug Fixes**
- Enhanced error handling for invalid stream names during ingestion,
ensuring appropriate logging and response for empty or invalid entries.
- Updated assertions in tests to reflect the correct handling of invalid
bulk payloads by the API, ensuring errors are present for invalid
submissions.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: hengfei yang <hengfei.yang@gmail.com>
---
 src/service/logs/bulk.rs             | 27 ++++++++++++++++++++++-
 tests/api-testing/tests/test_bulk.py | 33 +++-------------------------
 2 files changed, 29 insertions(+), 31 deletions(-)

diff --git a/src/service/logs/bulk.rs b/src/service/logs/bulk.rs
index 25856266fe..b9d71121ed 100644
--- a/src/service/logs/bulk.rs
+++ b/src/service/logs/bulk.rs
@@ -104,6 +104,31 @@ pub async fn ingest(
             }
             (action, stream_name, doc_id) = ret.unwrap();
 
+            if stream_name.is_empty() || stream_name == "_" || stream_name == "/" {
+                let err_msg = format!("Invalid stream name: {}", line);
+                log::warn!("{}", err_msg);
+                bulk_res.errors = true;
+                let err = BulkResponseError::new(
+                    err_msg.to_string(),
+                    stream_name.clone(),
+                    err_msg,
+                    "0".to_string(),
+                );
+                let mut item = HashMap::new();
+                item.insert(
+                    action.clone(),
+                    BulkResponseItem::new_failed(
+                        stream_name.clone(),
+                        doc_id.clone().unwrap_or_default(),
+                        err,
+                        Some(value),
+                        stream_name.clone(),
+                    ),
+                );
+                bulk_res.items.push(item);
+                continue; // skip
+            }
+
             if !cfg.common.skip_formatting_stream_name {
                 stream_name = format_stream_name(&stream_name);
             }
@@ -534,7 +559,7 @@ pub fn add_record_status(
                 failure_type,
                 stream_name.clone(),
                 failure_reason.unwrap(),
-                "0".to_owned(), // TODO check
+                "0".to_owned(),
             );
 
             item.insert(
diff --git a/tests/api-testing/tests/test_bulk.py b/tests/api-testing/tests/test_bulk.py
index cbdec1a6fe..87d2418300 100644
--- a/tests/api-testing/tests/test_bulk.py
+++ b/tests/api-testing/tests/test_bulk.py
@@ -30,33 +30,6 @@ def test_e2e_bulk_ingest(create_session, base_url):
         f"Expected no errors in the valid bulk request response, but got {resp_valid_bulk.json()}"
     )
 
-    # **Negative Test Case 1: Invalid Bulk Payload (Missing Document Body)**
-    invalid_bulk_payload = (
-        '{ "index" : { "_index" : "pytest10222", "_id" : "1" } }\n'
-        '{ "field1" : "value1" }'  # Add a document body after the index action
-    )
-
-    resp_invalid_bulk_1 = session.post(url, data=invalid_bulk_payload, headers=headers)
-    
-    # Print the raw response and content for debugging
-    print("Negative Test Response (Raw Content):", resp_invalid_bulk_1.content)
-    
-    try:
-        response_json = resp_invalid_bulk_1.json()  # Attempt to parse the JSON response
-        print("Negative Test Response (JSON):", response_json)
-    except Exception as e:
-        print("Error parsing JSON response:", str(e))
-
-    # Assert negative case for the first invalid payload
-    # Expecting a 200 response even for invalid data, but check that there's no error field
-    assert resp_invalid_bulk_1.status_code == 200, (
-        f"Expected status code 200 for invalid bulk request, but got {resp_invalid_bulk_1.status_code}. "
-        f"Response: {resp_invalid_bulk_1.content}"
-    )
-    assert not resp_invalid_bulk_1.json().get("errors"), (
-        f"Expected no errors in the invalid bulk request response, but got {resp_invalid_bulk_1.json()}"
-    )
-
     # **Negative Test Case 2: Invalid Bulk Payload (Empty Index)**
     invalid_bulk_payload2 = [
         { "index" : { "_index" : "" } },  # Empty index
@@ -85,9 +58,9 @@ def test_e2e_bulk_ingest(create_session, base_url):
         f"Response: {resp_invalid_bulk_2.content}"
     )
 
-    # Assert no errors in the response, as expected from Postman behavior
-    assert not resp_invalid_bulk_2.json().get("errors"), (
-        f"Expected no errors in the invalid bulk request response, but got {resp_invalid_bulk_2.json()}"
+    # Assert errors is true in the response
+    assert resp_invalid_bulk_2.json().get("errors"), (
+        f"Expected errors to be true in the invalid bulk request response, but got {resp_invalid_bulk_2.json()}"
     )