fix: flattening for kinesis #1329
Conversation
current: server expects decoded data to be a json object and adds timestamp and recordId to each json object change: the decoded data can be either a json object or a heavily nested json server decodes the data, flattens it (using generic flattening), then adds timestamp and recordId to each flattened json object also, all flattened json objects are converted as json array then ingested using common ingestion flow
WalkthroughThe changes refactor how Kinesis log messages are processed and flattened. The flattening function for Kinesis logs is now asynchronous, performs additional JSON structure checks, and flattens nested data. The ingestion utility now awaits this async function and pushes all flattened logs in a single batch. Flattening logic for Kinesis in the general JSON utility was also removed. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant IngestUtils
participant KinesisHandler
participant JSONUtils
Client->>IngestUtils: flatten_and_push_logs(LogSource::Kinesis, message)
IngestUtils->>KinesisHandler: await flatten_kinesis_logs(message)
KinesisHandler->>JSONUtils: has_more_than_max_allowed_levels(json)
KinesisHandler->>JSONUtils: generic_flattening(json)
KinesisHandler-->>IngestUtils: Result<Vec<Value>, Error>
IngestUtils->>IngestUtils: convert_to_array(flattened_logs)
IngestUtils->>IngestUtils: push_logs(array)
Poem
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
src/handlers/http/kinesis.rs (1)
67-89: Consider logging failed record processing.The current implementation silently skips records that fail UTF-8 conversion or exceed nesting limits. Consider adding logging for these cases to aid in debugging and monitoring.
if let Ok(json_string) = String::from_utf8(bytes) { let json: serde_json::Value = serde_json::from_str(&json_string)?; if !has_more_than_max_allowed_levels(&json, 1) { // ... existing flattening logic + } else { + tracing::warn!("Skipping record due to excessive nesting levels"); } + } else { + tracing::warn!("Failed to convert record data to UTF-8 string"); }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/handlers/http/kinesis.rs(2 hunks)src/handlers/http/modal/utils/ingest_utils.rs(1 hunks)src/utils/json/mod.rs(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: coverage
🔇 Additional comments (5)
src/utils/json/mod.rs (1)
47-48: Good separation of concerns for Kinesis log handling.Removing
LogSource::Kinesisfrom the generic flattening condition is appropriate since Kinesis logs now have dedicated async flattening logic insrc/handlers/http/kinesis.rs. This change aligns well with the PR objectives and maintains clean separation between generic and source-specific processing.src/handlers/http/modal/utils/ingest_utils.rs (1)
64-66: Excellent improvement to use async batch processing.The change from synchronous iteration to awaiting the async
flatten_kinesis_logsfunction and pushing the entire batch at once is a significant improvement. This approach is more efficient and aligns well with the new asynchronous flattening implementation.src/handlers/http/kinesis.rs (3)
62-62: Good improvement making the function async.Converting to an async function with proper Result return type aligns well with the architecture and enables better error handling. The return type change to
Result<Vec<Value>, anyhow::Error>is appropriate.
24-24: Good addition of flattening utilities.Adding the necessary imports for
generic_flatteningandhas_more_than_max_allowed_levelsenables the improved flattening logic.
69-70: Good optimization with nesting level check.Adding the nesting level check before applying generic flattening is a smart optimization that prevents unnecessary processing of deeply nested structures.
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
src/handlers/http/kinesis.rs (2)
66-68: Excellent! Past review comments have been addressed.The unwrap() calls have been properly replaced with the
?operator for error propagation, eliminating potential panics and providing proper error handling.
73-88: Well-implemented conditional flattening logic.The implementation correctly:
- Checks nesting levels before flattening
- Uses generic flattening for complex structures
- Properly enriches each flattened object with requestId and timestamp
- Uses proper error propagation instead of panic (addressing past review comments)
This addresses the core PR objective of handling heavily nested JSON structures.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/handlers/http/kinesis.rs(2 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
src/handlers/http/kinesis.rs (1)
src/utils/json/flatten.rs (3)
flatten(58-93)generic_flattening(269-328)has_more_than_max_allowed_levels(335-348)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: coverage
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
🔇 Additional comments (4)
src/handlers/http/kinesis.rs (4)
24-25: LGTM! Appropriate imports for new flattening functionality.The imports for
generic_flatteningandhas_more_than_max_allowed_levelsare correctly added to support the new conditional flattening logic.
62-62: Good improvement to async function with proper error handling.The function signature change to async with
Result<Vec<Value>, anyhow::Error>return type enables better error handling and integration with async processing flows. This aligns well with the PR objectives.
99-110: Proper error handling for UTF-8 conversion failures.The error logging and anyhow error return provide clear context about the failure, including requestId and timestamp for debugging. This is a significant improvement over the previous implementation.
113-113: Correct return statement for the new function signature.The
Ok(vec_kinesis_json)properly wraps the result vector in the Result type as expected by the new function signature.
| } else { | ||
| // If the JSON has more than the allowed levels, we just push it as is | ||
| // without flattening or modifying it. | ||
| // This is a fallback to ensure we don't lose data. | ||
| tracing::warn!( | ||
| "Kinesis log with requestId {} and timestamp {} has more than the allowed levels of nesting, skipping flattening for this record.", | ||
| message.request_id, message.timestamp | ||
| ); | ||
| vec_kinesis_json.push(json); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider adding metadata consistency for deeply nested records.
When skipping flattening due to excessive nesting, the original JSON is pushed without adding requestId and timestamp fields. This creates inconsistent data structure compared to flattened records.
Consider adding the metadata even for non-flattened records:
} else {
// If the JSON has more than the allowed levels, we just push it as is
// without flattening or modifying it.
// This is a fallback to ensure we don't lose data.
+ let mut kinesis_json = json.as_object().unwrap_or(&Map::new()).clone();
+ kinesis_json.insert("requestId".to_owned(), Value::String(message.request_id.clone()));
+ kinesis_json.insert("timestamp".to_owned(), Value::String(message.timestamp.to_string()));
tracing::warn!(
"Kinesis log with requestId {} and timestamp {} has more than the allowed levels of nesting, skipping flattening for this record.",
message.request_id, message.timestamp
);
- vec_kinesis_json.push(json);
+ vec_kinesis_json.push(Value::Object(kinesis_json));
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| } else { | |
| // If the JSON has more than the allowed levels, we just push it as is | |
| // without flattening or modifying it. | |
| // This is a fallback to ensure we don't lose data. | |
| tracing::warn!( | |
| "Kinesis log with requestId {} and timestamp {} has more than the allowed levels of nesting, skipping flattening for this record.", | |
| message.request_id, message.timestamp | |
| ); | |
| vec_kinesis_json.push(json); | |
| } | |
| } else { | |
| // If the JSON has more than the allowed levels, we just push it as is | |
| // without flattening or modifying it. | |
| // This is a fallback to ensure we don't lose data. | |
| let mut kinesis_json = json.as_object().unwrap_or(&Map::new()).clone(); | |
| kinesis_json.insert( | |
| "requestId".to_owned(), | |
| Value::String(message.request_id.clone()), | |
| ); | |
| kinesis_json.insert( | |
| "timestamp".to_owned(), | |
| Value::String(message.timestamp.to_string()), | |
| ); | |
| tracing::warn!( | |
| "Kinesis log with requestId {} and timestamp {} has more than the allowed levels of nesting, skipping flattening for this record.", | |
| message.request_id, message.timestamp | |
| ); | |
| vec_kinesis_json.push(Value::Object(kinesis_json)); | |
| } |
🤖 Prompt for AI Agents
In src/handlers/http/kinesis.rs around lines 89 to 98, when skipping flattening
for deeply nested JSON records, the code pushes the original JSON without adding
the requestId and timestamp metadata fields. To maintain consistent data
structure, modify the code to insert the requestId and timestamp into the JSON
object before pushing it to vec_kinesis_json, ensuring all records include these
metadata fields regardless of flattening.
There was a problem hiding this comment.
if the decoded json is heavily nested and it's a json array, unwrapping it as object will always fail and unwrap_or(&Map::new()) will end up in data loss, hence pushing the actual nested json to process is the best we can do here. Obviously, we cannot add metadata to the json array unless flattened, hence, we will have to push the original json as is for processing. Hope the description clarifies the changes.
There was a problem hiding this comment.
@nikhilsinhaparseable Thank you for the detailed explanation! You're absolutely right. I didn't consider the case where the decoded JSON could be an array at the root level. Your analysis is spot-on:
- For JSON arrays,
json.as_object()would returnNone - Using
unwrap_or(&Map::new())would indeed cause data loss by replacing the array with an empty object - Metadata fields cannot be directly added to JSON arrays - they need to be flattened first to become individual objects
- When flattening is skipped due to nesting depth, preserving the original structure is the only viable option
The current implementation correctly handles this edge case by pushing the original JSON as-is. I appreciate the clarification!
✏️ Learnings added
Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1329
File: src/handlers/http/kinesis.rs:89-98
Timestamp: 2025-05-28T04:45:40.846Z
Learning: In Kinesis log processing, when handling deeply nested JSON that exceeds flattening limits, the original JSON structure must be preserved as-is. If the JSON is an array at the root level, attempting to add metadata fields like `requestId` and `timestamp` would fail because arrays don't support direct key-value insertion. The flattening process handles arrays by converting elements to individual objects, but when flattening is skipped, the original array structure must remain intact to prevent data loss.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
current: server expects decoded data to be a json object
and adds timestamp and recordId to each json object
change: the decoded data can be either a json object or a heavily nested json
server decodes the data, flattens it (using generic flattening),
then adds timestamp and recordId to each flattened json object
also, all flattened json objects are converted as json array
then ingested using common ingestion flow
Summary by CodeRabbit