starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 08:09:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: check invalid stream_name during bulk ingestion (#5403)

Browse Source 
#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>
This commit is contained in:
Loakesh Indiran
2024-12-18 18:08:31 +05:30
committed by GitHub
co-authored by hengfei yang
parent 16fb8eb3a4
commit 8d234483d2
2 changed files with 29 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/service/logs/bulk.rs
+26 -1
View File
@@ -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(
tests/api-testing/tests/test_bulk.py
+3 -30
View File
@@ -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()}"
)