refactor: move kinesis to src/
#1269
Conversation
WalkthroughThis pull request removes the legacy Kinesis log parsing functionality from the HTTP handlers and reintroduces it in a new module. The functionality originally in Changes
Sequence Diagram(s)sequenceDiagram
participant C as Caller
participant F as flatten_kinesis_logs
participant B as Base64 Decoder
participant J as JSON Deserializer
C->>F: Call flatten_kinesis_logs(Message)
F->>F: Iterate over each record
F->>B: Decode base64 data
B-->>F: Decoded string
F->>J: Deserialize JSON
J-->>F: JSON object (or error)
F->>F: Append request_id & timestamp
F-->>C: Return vector of JSON objects
Poem
✨ Finishing Touches
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
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.
Caution
Inline review comments failed to post. This is likely due to GitHub's limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 1
🧹 Nitpick comments (2)
src/kinesis.rs (2)
24-34: Validate naming of structs and ensure future extensibility.The
Messagestruct and the nestedDatastruct are named generically. If there's any chance that these data structures evolve to handle more Kinesis event fields (e.g. metadata), consider giving them more descriptive names or making them more flexible. Otherwise, they look fine for the current scope.
87-227: Great coverage of typical scenarios.The tests thoroughly validate typical Kinesis logs, successfully confirming base64 decoding, correct JSON structure, and proper handling of malformed data. You might want to add a test for an empty
recordsarray to ensure correct behavior in that edge case.
🛑 Comments failed to post (1)
src/kinesis.rs (1)
36-85: 🛠️ Refactor suggestion
Avoid panicking on invalid data.
Calling
unwrap()on base64 decoding and JSON parsing causes a full panic if data is malformed. While there’s a test that verifies this behavior, you might consider returning an error or skipping malformed records rather than bringing down the entire server in production.-pub fn flatten_kinesis_logs(message: Message) -> Vec<Value> { +pub fn flatten_kinesis_logs(message: Message) -> Result<Vec<Value>, SomeErrorType> { ... - let bytes = STANDARD.decode(record.data.clone()).unwrap(); + let bytes = STANDARD.decode(record.data.clone()) + .map_err(|e| SomeErrorType::DecodeError(e.to_string()))?; ... - let json_string: String = String::from_utf8(bytes).unwrap(); + let json_string: String = String::from_utf8(bytes) + .map_err(|e| SomeErrorType::Utf8Error(e.to_string()))?; ... - let json: serde_json::Value = serde_json::from_str(&json_string).unwrap(); + let json: serde_json::Value = serde_json::from_str(&json_string) + .map_err(|e| SomeErrorType::SerdeError(e.to_string()))?; ... - Err(error) => panic!("Failed to deserialize JSON: {}", error), + Err(error) => return Err(SomeErrorType::JsonDeserialization(error.to_string())), ... }
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
src/kinesis.rs (2)
183-196: Consider adding more error test casesWhile you have a test for invalid base64 data, consider adding tests for other error scenarios like malformed JSON after decoding valid base64, or empty records array.
If you implement the error handling suggestion above, you should update this test to verify errors are returned rather than expecting panics:
#[test] fn malformed_json_after_base64_decoding_returns_error() { let message: Message = serde_json::from_value(json!({ "requestId": "9b848d8a-2d89-474b-b073-04b8e5232210".to_string(), "timestamp": 1705026780451_i64, "records": [ { "data": "invalid_base64_data".to_string(), }], })) .unwrap(); let result = flatten_kinesis_logs(message); assert!(result.is_err()); }
72-79: Use consistent string conversion methodsThe code uses both
to_owned()andto_string()for similar string conversions. Consider standardizing on one approach for consistency.kinesis_json.insert( "requestId".to_owned(), Value::String(message.request_id.clone()), ); kinesis_json.insert( - "timestamp".to_owned(), + "timestamp".to_string(), Value::String(message.timestamp.to_string()), );Or alternatively:
kinesis_json.insert( - "requestId".to_owned(), + "requestId".to_string(), Value::String(message.request_id.clone()), ); kinesis_json.insert( - "timestamp".to_owned(), + "timestamp".to_string(), Value::String(message.timestamp.to_string()), );
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/kinesis.rs(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- 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: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: coverage
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
🔇 Additional comments (4)
src/kinesis.rs (4)
24-34: Good structure definition for Kinesis message formatThe
MessageandDatastructs are well-defined with appropriate serialization attributes. The#[serde(rename_all = "camelCase")]attribute ensures proper JSON mapping.
36-59: Good documentation with clear format examplesThe comments provide excellent explanations of the input and output formats with concrete examples, making the code self-documenting and easy to understand.
87-142: Comprehensive test coverage for base64 decoding functionalityThe test case thoroughly validates the flattening process, ensuring that base64 data is correctly decoded and all expected fields are present with the right values.
144-181: Good test for metadata field additionThis test properly verifies that the requestId and timestamp are correctly added to each record in the output.
| pub fn flatten_kinesis_logs(message: Message) -> Vec<Value> { | ||
| let mut vec_kinesis_json = Vec::new(); | ||
|
|
||
| for record in message.records.iter() { | ||
| let bytes = STANDARD.decode(record.data.clone()).unwrap(); | ||
| let json_string: String = String::from_utf8(bytes).unwrap(); | ||
| let json: serde_json::Value = serde_json::from_str(&json_string).unwrap(); | ||
| let mut kinesis_json: Map<String, Value> = match serde_json::from_value(json) { | ||
| Ok(value) => value, | ||
| Err(error) => panic!("Failed to deserialize JSON: {}", error), | ||
| }; | ||
|
|
||
| kinesis_json.insert( | ||
| "requestId".to_owned(), | ||
| Value::String(message.request_id.clone()), | ||
| ); | ||
| kinesis_json.insert( | ||
| "timestamp".to_owned(), | ||
| Value::String(message.timestamp.to_string()), | ||
| ); | ||
|
|
||
| vec_kinesis_json.push(Value::Object(kinesis_json)); | ||
| } | ||
|
|
||
| vec_kinesis_json | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Improve error handling to avoid potential panics
The function uses multiple unwrap() calls which will cause the program to panic if errors occur. Consider using proper error handling with Result types instead.
-pub fn flatten_kinesis_logs(message: Message) -> Vec<Value> {
+pub fn flatten_kinesis_logs(message: Message) -> Result<Vec<Value>, Box<dyn std::error::Error>> {
let mut vec_kinesis_json = Vec::new();
for record in message.records.iter() {
- let bytes = STANDARD.decode(record.data.clone()).unwrap();
- let json_string: String = String::from_utf8(bytes).unwrap();
- let json: serde_json::Value = serde_json::from_str(&json_string).unwrap();
- let mut kinesis_json: Map<String, Value> = match serde_json::from_value(json) {
- Ok(value) => value,
- Err(error) => panic!("Failed to deserialize JSON: {}", error),
- };
+ let bytes = STANDARD.decode(&record.data)?;
+ let json_string = String::from_utf8(bytes)?;
+ let json: serde_json::Value = serde_json::from_str(&json_string)?;
+ let mut kinesis_json: Map<String, Value> = serde_json::from_value(json)?;
kinesis_json.insert(
"requestId".to_owned(),
- Value::String(message.request_id.clone()),
+ Value::String(message.request_id.to_owned()),
);
kinesis_json.insert(
"timestamp".to_owned(),
Value::String(message.timestamp.to_string()),
);
vec_kinesis_json.push(Value::Object(kinesis_json));
}
- vec_kinesis_json
+ Ok(vec_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.
| pub fn flatten_kinesis_logs(message: Message) -> Vec<Value> { | |
| let mut vec_kinesis_json = Vec::new(); | |
| for record in message.records.iter() { | |
| let bytes = STANDARD.decode(record.data.clone()).unwrap(); | |
| let json_string: String = String::from_utf8(bytes).unwrap(); | |
| let json: serde_json::Value = serde_json::from_str(&json_string).unwrap(); | |
| let mut kinesis_json: Map<String, Value> = match serde_json::from_value(json) { | |
| Ok(value) => value, | |
| Err(error) => panic!("Failed to deserialize JSON: {}", error), | |
| }; | |
| kinesis_json.insert( | |
| "requestId".to_owned(), | |
| Value::String(message.request_id.clone()), | |
| ); | |
| kinesis_json.insert( | |
| "timestamp".to_owned(), | |
| Value::String(message.timestamp.to_string()), | |
| ); | |
| vec_kinesis_json.push(Value::Object(kinesis_json)); | |
| } | |
| vec_kinesis_json | |
| } | |
| pub fn flatten_kinesis_logs(message: Message) -> Result<Vec<Value>, Box<dyn std::error::Error>> { | |
| let mut vec_kinesis_json = Vec::new(); | |
| for record in message.records.iter() { | |
| let bytes = STANDARD.decode(&record.data)?; | |
| let json_string = String::from_utf8(bytes)?; | |
| let json: serde_json::Value = serde_json::from_str(&json_string)?; | |
| let mut kinesis_json: Map<String, Value> = serde_json::from_value(json)?; | |
| kinesis_json.insert( | |
| "requestId".to_owned(), | |
| Value::String(message.request_id.to_owned()), | |
| ); | |
| kinesis_json.insert( | |
| "timestamp".to_owned(), | |
| Value::String(message.timestamp.to_string()), | |
| ); | |
| vec_kinesis_json.push(Value::Object(kinesis_json)); | |
| } | |
| Ok(vec_kinesis_json) | |
| } |
Fixes #XXXX.
Description
src/kinesis.rsThis PR has:
Summary by CodeRabbit