fix: parallelise datasets API #1276
Conversation
process each stream in parallel process distinct counts in parallel
WalkthroughThis change refactors the Changes
Possibly related PRs
Suggested labels
Suggested reviewers
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.
Actionable comments posted: 1
🧹 Nitpick comments (6)
src/prism/logstream/mod.rs (6)
249-249: Consider concurrency limits for large stream lists.If the system has a large number of streams, listing them all and processing concurrently can overwhelm resources. You may consider chunking or otherwise bounding concurrency to prevent resource exhaustion.
253-259: Use bounded concurrency when needed.
futures::future::join_allruns all tasks at once. Ifself.streamsgrows large, the system may spawn excessive concurrent tasks. Consider using a method such asfutures::stream::iter(...).buffer_unordered(...)to cap concurrency.- let results = futures::future::join_all( - self.streams - .iter() - .map(|stream| self.process_stream(stream.clone(), key.clone())), - ).await; + let results = futures::stream::iter( + self.streams.iter().map(|stream| { + self.process_stream(stream.clone(), key.clone()) + }) + ) + .buffer_unordered(10) // or any appropriate limit + .collect::<Vec<_>>() + .await;
273-294: Simplify function signature for clarity.
process_streamreturnsOption<Result<...>>, which can be confusing. A more direct approach might be returningResult<Option<PrismDatasetResponse>, PrismLogstreamError>to unify error handling and optional logic.- async fn process_stream(&self, stream: String, key: SessionKey) - -> Option<Result<PrismDatasetResponse, PrismLogstreamError>> { + async fn process_stream(&self, stream: String, key: SessionKey) + -> Result<Option<PrismDatasetResponse>, PrismLogstreamError> { // Then return Ok(None) if unauthorized or non-existent, // and return Ok(Some(...)) or Err(...) otherwise. }
307-314: Clarify the meaning ofcheck_or_load_streamreturn value.Currently,
check_or_load_stream(stream)returningtrueindicates the stream does not exist, making the method returnfalseforstream_exists. The naming is counterintuitive. Consider renaming to something likeis_stream_missing(...)or reversing its return value for clarity.
316-349: Handle distinct entry retrieval errors more visibly.Lines 333–334 quietly ignore errors (using
.ok()) for IPs/user_agents retrieval. Consider logging or returning partial errors so that administrators can diagnose query failures for distinct fields.
367-380: Make hardcoded time range and bin count configurable.Currently, the time range is fixed at "1h" to "now" with 10 bins (lines 370–373). Consider allowing clients to specify these parameters to accommodate various query requirements and reduce potential overhead from unneeded bins.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/prism/logstream/mod.rs(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: coverage
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
🔇 Additional comments (2)
src/prism/logstream/mod.rs (2)
296-305: Authorization check looks good.This function cleanly logs unauthorized attempts without exposing sensitive details. No issues found here.
351-365: Global manager usage is acceptable.
get_hot_tier_inforesponsibly handles missing or invalid hot tier data, returningNoneor the error. The concurrency aspects here look fine.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
src/prism/logstream/mod.rs (3)
264-273: Consider adding stream context to error logs.When logging errors, it would be helpful to include the stream name for better debugging context.
- Err(err) => { - warn!("error: {err}"); - continue; + Err(err) => { + warn!("Error processing stream '{}': {}", stream, err); + continue; }
313-320: Consider adjusting log level for stream existence checks.For cases where you're querying all streams and some might legitimately not exist, a warning log level might be too high. Consider using
debuglevel instead, especially if this is an expected condition in some scenarios.- warn!("Stream not found: {stream}"); + debug!("Stream not found: {stream}");
373-386: Document or parameterize the hardcoded time values.The
get_countsmethod uses hardcoded values forstart_time,end_time, andnum_bins. Consider making these configurable parameters or documenting why these specific values are used.async fn get_counts(&self, stream: &str) -> Result<CountsResponse, PrismLogstreamError> { + // Using last hour of data with 10 bins as default for dashboard visualization let count_request = CountsRequest { stream: stream.to_owned(), start_time: "1h".to_owned(), end_time: "now".to_owned(), num_bins: 10, };Alternatively, consider parameterizing these values:
- async fn get_counts(&self, stream: &str) -> Result<CountsResponse, PrismLogstreamError> { + async fn get_counts( + &self, + stream: &str, + start_time: Option<String>, + end_time: Option<String>, + num_bins: Option<usize> + ) -> Result<CountsResponse, PrismLogstreamError> { let count_request = CountsRequest { stream: stream.to_owned(), - start_time: "1h".to_owned(), - end_time: "now".to_owned(), - num_bins: 10, + start_time: start_time.unwrap_or_else(|| "1h".to_owned()), + end_time: end_time.unwrap_or_else(|| "now".to_owned()), + num_bins: num_bins.unwrap_or(10), };
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/prism/logstream/mod.rs(2 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 x86_64-apple-darwin
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: coverage
🔇 Additional comments (6)
src/prism/logstream/mod.rs (6)
27-27: Good addition of the warn module.Adding the
warnmodule fromtracingis appropriate for the improved error handling implemented in this PR.
249-277: Great implementation of concurrent stream processing!This refactoring successfully implements parallel processing of streams using
futures::future::join_all, which aligns perfectly with the PR objective to enhance performance. The error handling has also been significantly improved - now logging errors and continuing processing instead of failing the entire operation.This change addresses the past review comment about inconsistent error handling.
280-300: Well-structured stream processing method.Good implementation of the
process_streammethod with clear separation of concerns - checking authorization, verifying stream existence, and then processing the stream data. The early returns for unauthorized or non-existent streams prevent unnecessary processing.
302-311: Clean authorization check implementation.The
is_authorizedmethod properly encapsulates the authorization logic with good error logging.
322-355: Excellent use of concurrent operations in response building.The parallel fetching of distinct entries using
futures::join!aligns well with the PR's objective of enhancing performance through parallelization. Also good error handling with.ok()on the results.
357-371: Clean implementation of hot tier info retrieval.Good error handling with proper pattern matching on different error types.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (5)
src/prism/logstream/mod.rs (5)
249-251: Clarify fallback logic whenstreamsis empty
Whenself.streamsis empty, it is replaced with the entire list of streams. This behavior may be surprising to users expecting an empty request to return nothing. Consider documenting or confirming this fallback with the caller to ensure it’s intentional.
280-300: Consider surfacing partial errors inprocess_stream
When a stream is missing or unauthorized, this function returnsOk(None). Users may benefit from more granular logging or a structured report (e.g., which streams failed). The current approach is fine for skipping failed streams, but consider whether partial errors should be communicated more explicitly to the end user.
302-311: Minimize cloning ofSessionKey
Theis_authorizedmethod clones the session key each time. Although cheap here, consider passing a reference to avoid unneeded copies if usage increases.
313-320: Rename or remove log message instream_exists
This method logs a warning if the stream is not found, yet it returnsfalseto indicate the same condition. If repeated calls happen for the same missing stream, you risk spamming logs. Consider promoting this warning to a debug-level log, or rename the function tocheck_stream_and_log_if_missingto clarify side effects.
373-385: Parameterize the time range
Theget_countsfunction uses a fixed start time of"1h"and end time of"now". If you anticipate different time windows from consumers, consider making these fields configurable withinPrismDatasetRequest.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/prism/logstream/mod.rs(2 hunks)
🧰 Additional context used
🪛 GitHub Actions: Lint, Test and Coverage Report
src/prism/logstream/mod.rs
[error] 271-271: error[E0425]: cannot find value stream in this scope
⏰ Context from checks skipped due to timeout of 90000ms (4)
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
🔇 Additional comments (2)
src/prism/logstream/mod.rs (2)
253-260: Good use of concurrent processing
Usingfutures::future::join_allfor concurrently processing each stream is a solid approach, ensuring the method won’t block on stream processing. The concurrency design aligns well with the goal of parallelizing dataset retrieval.
322-355: Parallel retrieval of distinct field values
Performing twoget_distinct_entriescalls withfutures::join!is efficient. Ensure that this does not introduce excessive overhead for large datasets. Optionally, you can unify or cache these calls if further expansions (e.g., multiple fields) are expected.
nikhilsinhaparseable
force-pushed
the
parallel-dataset
branch
from
d530b72 to
dc754d8
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (4)
src/prism/logstream/mod.rs (4)
250-252: Consider filtering out unauthorized streams earlier.
Currently, ifstreams.is_empty()we pull all streams, but only later skip unauthorized ones inprocess_stream(). Consider an early filter to avoid generating possibly large unnecessary futures for streams the user cannot access, improving overall performance.
254-261: Limit concurrency if the stream list is large.
Usingfutures::future::join_allwill spawn a task for each stream. If the user can query hundreds or thousands of streams at once, this could create significant overhead. Consider using a bounded concurrency approach likefutures::stream::iter(...).for_each_concurrent(...)with a reasonable limit to prevent resource exhaustion.
314-321: Clarify the return value ofstream_exists().
The function name suggests it returnstrueif the stream exists, but it returnsfalseifcheck_or_load_stream()istrue. This can be confusing. Consider renaming to something likestream_missing()or flipping the condition to improve readability.
374-387: Parameterize the time range if needed.
Currently, the start and end times for counts are hardcoded to1handnow. If the user or other components eventually require flexibility, consider exposing them as optional parameters for this method.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/prism/logstream/mod.rs(2 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (10)
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Default aarch64-apple-darwin
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Default x86_64-apple-darwin
- GitHub Check: coverage
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
🔇 Additional comments (4)
src/prism/logstream/mod.rs (4)
27-27: No issues noted on this import addition.
This addition ofuse tracing::warn;is consistent with the logging approach used throughout the file.
262-279: Implementation aligns with the documented behavior.
The loop collects only successful responses and continues after errors, matching the doc comment that individual stream failures won’t halt the entire process. Good job adhering to the requirement.
281-301: Overall concurrency design looks correct.
Each stream is processed in its own future, and errors are handled gracefully. This is consistent with the stated PR objective to parallelize dataset retrieval without blocking on individual stream failures.
303-312: Method naming and usage is intuitive.
is_authorizedaccurately encodes the RBAC check, returningfalsewhen authorization fails. The logging of unauthorized access is appropriate.
process each stream in parallel
process distinct counts in parallel
Summary by CodeRabbit
Refactor
New Features