feat: Ingestion traces#1629
Merged
nikhilsinhaparseable merged 10 commits intoparseablehq:mainfrom Apr 28, 2026
Merged
Conversation
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds OpenTelemetry/tracing crates and initializer; instruments ingestion, parsing, storage, sync, and utility hot paths with tracing spans; converts ingest push to synchronous batched flow; extends CLI/Actix server connection settings; refactors staging reader and flush/convert locking/flow. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant ActixServer as Actix Server
participant IngestUtils as Ingest Utils
participant StreamMgr as Stream / Storage
participant Telemetry as OTLP Exporter
Client->>ActixServer: HTTP ingest request
ActixServer->>IngestUtils: parse & flatten batch (span)
IngestUtils->>IngestUtils: infer schema & compute json_byte_size (span)
IngestUtils->>StreamMgr: push_logs (synchronous batched Value::Array)
StreamMgr->>StreamMgr: write mem/disk, schedule flush/convert (spans)
StreamMgr->>Telemetry: emit spans/metrics via tracing -> OTLP exporter
Telemetry-->>Telemetry: export spans to configured OTLP endpoint
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (2)
src/parseable/staging/reader.rs (1)
86-101: Add direct tests for the new forward merge path.
MergedRecordReader::merged_iteris now the path used during parquet conversion, but the tests in this file still only exercise_MergedReverseRecordReader/_get_reverse_reader. Please add coverage for multiple files, equal timestamps, and customtime_partitionso ordering regressions show up here instead of during conversion.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/staging/reader.rs` around lines 86 - 101, Tests currently only exercise the reverse merge path (_MergedReverseRecordReader / _get_reverse_reader) but not the new forward merge path MergedRecordReader::merged_iter used during parquet conversion; add unit tests that call merged_iter (via MergedRecordReader) to assert global ordering across multiple IPC files, include cases with equal timestamps to ensure stable ordering, and a case using a non-default time_partition string so get_timestamp_millis(time_partition) is exercised; create small in-memory RecordBatches with controlled timestamps and multiple readers, feed them into merged_iter, and assert the output sequence matches the expected ascending order and respects your time_partition extraction.src/handlers/http/modal/mod.rs (1)
132-139: Useinfo!for the startup config dump.This path runs on every healthy boot, so logging it at warning level will create noisy alerts for normal behavior.
Suggested change
- tracing::warn!( + info!( "Starting Query server with-\nNum workers: {}\nKeep Alive: {}\nRequest timeout: {}\nConnection backlog: {}\nMax connections: {}", PARSEABLE.options.num_workers, PARSEABLE.options.keep_alive, PARSEABLE.options.request_timeout, PARSEABLE.options.connection_backlog, PARSEABLE.options.max_connections );🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/handlers/http/modal/mod.rs` around lines 132 - 139, The startup configuration log currently uses tracing::warn! which is too noisy for normal healthy boots; change the call to tracing::info! so the configuration dump (the message using PARSEABLE.options.num_workers, .keep_alive, .request_timeout, .connection_backlog, .max_connections) is emitted at info level instead of warn, leaving the message text and interpolated fields unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/cli.rs`:
- Around line 169-176: The num_workers CLI field currently allows 0 which later
causes a panic when passed to HttpServer::workers(...); add a custom parser
function (e.g., parse_num_workers) that accepts a &str, parses to usize, and
returns an error if the value is 0, then reference it on the num_workers arg via
#[arg(value_parser = parse_num_workers)] so invalid values are rejected at
argument parsing time instead of panicking at HttpServer startup.
In `@src/handlers/http/modal/utils/ingest_utils.rs`:
- Around line 245-247: The code is converting the broken ControlFlow error to a
string and wrapping it in PostError::CustomError which loses the original error
type; instead preserve the error by returning it as a PostError using From/Into
or a dedicated variant: change the ControlFlow::Break(e) branch to return
Err(e.into()) (or Err(PostError::from(e))) if PostError implements From for the
error type, or add a new PostError variant like PostError::ParallelError(Box<dyn
std::error::Error + Send + Sync>) and return
Err(PostError::ParallelError(Box::new(e))) so the original error type/context is
retained (refer to ControlFlow::Break, r, and PostError::CustomError in the
code).
- Around line 188-191: The code unwraps data.into_iter().next() and can panic if
convert_array_to_object returns an empty Vec; update the block in
ingest_utils.rs around custom_partition/process_non_partitioned to handle an
empty iterator safely (e.g., check data.is_empty() or use .next().ok_or / match)
and return an appropriate error or empty result instead of unwrapping; ensure
the replacement sets json_batch only when an element exists and preserves the
existing behavior when data has one element.
- Around line 153-158: The tracing span for push_logs declares a field
record_count but never sets it; update the push_logs function to call
Span::current().record("record_count", &tracing::field::display(n)) (or use
tracing::Span::current().record with the actual numeric count) at the point
where you know how many records were processed so the field is populated; locate
the push_logs function and add the record call after computing the record count
(use the same "record_count" name and a Display-able numeric value) so the
instrumentation reflects the real value.
In `@src/otel/metrics.rs`:
- Around line 457-464: The empty-data_points fallback currently builds a
synthetic Map with only generic fields (metric_name, metric_description,
metric_unit, metric_type_val, metadata) and thus drops type-specific fields like
is_monotonic and aggregation_temporality; update the fallback in the branch that
constructs single (the Map created with Map::new()) to also insert the same
type-specific entries the non-empty path uses (e.g., insert "is_monotonic" when
the metric is a Sum and insert "aggregation_temporality" for sums/histograms) by
reusing the existing variables/values (is_monotonic, aggregation_temporality or
whatever names are in scope) so the synthetic record matches the non-empty
record shape.
In `@src/parseable/streams.rs`:
- Around line 688-691: The match on MergedRecordReader::try_new(&arrow_files)
can return a reader with reader.readers.is_empty(), and the current `_ =>
continue` leaves the moved files under the processing_* directory with no
cleanup; change this branch to call the same cleanup/rollback logic you use for
successful processing (e.g., remove or move out the processing_* directories for
the given arrow_files) before continuing. Specifically, in the branch where
MergedRecordReader::try_new returns Err or Ok(reader) with
reader.readers.is_empty(), invoke the cleanup routine that cleans up the
arrow_files / processing_* group (the same code path used after successful
conversion) instead of simply continuing, so processing_* directories don’t
accumulate; reference MergedRecordReader::try_new, the record_reader variable,
and the arrow_files/processing_* group when locating where to insert the cleanup
call.
In `@src/sync.rs`:
- Around line 134-156: The REMOTE_SYNC_RUNNING flag is only cleared on the happy
path, so if anything inside the async block panics or returns early the flag
remains true; wrap the compare_exchange and subsequent work in a RAII drop guard
that sets REMOTE_SYNC_RUNNING.store(false, Ordering::SeqCst) in its Drop impl
(or use scopeguard) so the flag is always cleared on exit, and apply the same
pattern to LOCAL_SYNC_RUNNING and the other similar sync blocks (see the
identical logic around lines referenced in the review) that call
sync_all_streams/monitor_task_duration to ensure all exit paths clear their
guard.
In `@src/telemetry.rs`:
- Around line 48-56: The early return in init_tracing() only checks
OTEL_EXPORTER_OTLP_ENDPOINT, incorrectly disabling tracing when only
OTEL_EXPORTER_OTLP_TRACES_ENDPOINT is set; change the existence check to succeed
if either std::env::var(OTEL_EXPORTER_OTLP_ENDPOINT).ok() or
std::env::var(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT).ok() is present so
signal-specific traces configuration enables tracing; keep the rest of
init_tracing() (including reading OTEL_EXPORTER_OTLP_PROTOCOL) unchanged.
---
Nitpick comments:
In `@src/handlers/http/modal/mod.rs`:
- Around line 132-139: The startup configuration log currently uses
tracing::warn! which is too noisy for normal healthy boots; change the call to
tracing::info! so the configuration dump (the message using
PARSEABLE.options.num_workers, .keep_alive, .request_timeout,
.connection_backlog, .max_connections) is emitted at info level instead of warn,
leaving the message text and interpolated fields unchanged.
In `@src/parseable/staging/reader.rs`:
- Around line 86-101: Tests currently only exercise the reverse merge path
(_MergedReverseRecordReader / _get_reverse_reader) but not the new forward merge
path MergedRecordReader::merged_iter used during parquet conversion; add unit
tests that call merged_iter (via MergedRecordReader) to assert global ordering
across multiple IPC files, include cases with equal timestamps to ensure stable
ordering, and a case using a non-default time_partition string so
get_timestamp_millis(time_partition) is exercised; create small in-memory
RecordBatches with controlled timestamps and multiple readers, feed them into
merged_iter, and assert the output sequence matches the expected ascending order
and respects your time_partition extraction.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 255d8234-a2d6-4ff7-a902-148688550cd3
📒 Files selected for processing (16)
Cargo.tomlsrc/cli.rssrc/event/format/json.rssrc/event/format/mod.rssrc/event/mod.rssrc/handlers/http/modal/mod.rssrc/handlers/http/modal/utils/ingest_utils.rssrc/lib.rssrc/otel/metrics.rssrc/parseable/staging/reader.rssrc/parseable/streams.rssrc/storage/field_stats.rssrc/storage/object_storage.rssrc/sync.rssrc/telemetry.rssrc/utils/json/mod.rs
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
src/parseable/staging/reader.rs (1)
147-147: Consider using#[allow(dead_code)]instead of underscore prefix for unused functions.The underscore-prefix naming convention (
_new,_find_limit_and_type) is typically used for unused variables in Rust, not functions. If these are internal-only and the underscore is to suppress unused warnings, consider using#[allow(dead_code)]or#[cfg(test)]attributes instead, which convey intent more clearly.Also applies to: 246-246
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/staging/reader.rs` at line 147, The underscore-prefixed function names like _new and _find_limit_and_type should not be used to silence unused warnings; instead remove the leading underscore and mark the internal-only functions with #[allow(dead_code)] (or #[cfg(test)] if they are only used in tests). Locate the functions named _new and _find_limit_and_type and replace the underscore naming with normal names (new, find_limit_and_type) and add #[allow(dead_code)] above each function (or switch to #[cfg(test)] when appropriate) so intent is explicit and warnings are suppressed correctly.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/parseable/staging/reader.rs`:
- Around line 516-517: Tests call non-existent underscored constructors/methods
for MergedReverseRecordReader; change calls from
MergedReverseRecordReader::_try_new(...) and ._merged_iter(...) to the actual
public names MergedReverseRecordReader::try_new(...) and .merged_iter(...);
apply this same replacement for the second occurrence of the same calls later in
the file (the other MergedReverseRecordReader::_try_new/_merged_iter usage).
- Line 603: Rename the misnamed test function
testget_reverse_reader_single_message to follow snake_case:
test_get_reverse_reader_single_message; update any references or attributes
(e.g., the #[test] function declaration) that call or annotate this function to
use the new name so the test runs correctly and to match Rust naming
conventions.
---
Nitpick comments:
In `@src/parseable/staging/reader.rs`:
- Line 147: The underscore-prefixed function names like _new and
_find_limit_and_type should not be used to silence unused warnings; instead
remove the leading underscore and mark the internal-only functions with
#[allow(dead_code)] (or #[cfg(test)] if they are only used in tests). Locate the
functions named _new and _find_limit_and_type and replace the underscore naming
with normal names (new, find_limit_and_type) and add #[allow(dead_code)] above
each function (or switch to #[cfg(test)] when appropriate) so intent is explicit
and warnings are suppressed correctly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 454be6bf-abb5-47c7-873a-8cfe93d88955
📒 Files selected for processing (2)
src/parseable/staging/reader.rssrc/parseable/streams.rs
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
src/parseable/staging/reader.rs (1)
602-602:⚠️ Potential issue | 🟡 MinorRename test to snake_case (
test_get_reverse_reader_single_message).Line 602 still has the previously reported typo
testget_reverse_reader_single_message.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/staging/reader.rs` at line 602, Rename the test function symbol testget_reverse_reader_single_message to use snake_case (test_get_reverse_reader_single_message) so it follows Rust naming conventions; update the function declaration fn testget_reverse_reader_single_message() -> io::Result<()> to fn test_get_reverse_reader_single_message() -> io::Result<()> and update any references to that function within the file (e.g., test registration or calls) to the new name.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/parseable/staging/reader.rs`:
- Line 241: The code calls StreamReader::try_new(...).unwrap(), which panics on
Arrow IPC/stream parsing errors and breaks the Result-based error flow; replace
the unwrap by propagating the error through the function's Result (e.g., use ?
or map_err to convert the Arrow error into the crate's error type) so
StreamReader::try_new(BufReader::new(OffsetReader::_new(reader, messages)),
None) returns an Err instead of panicking and the caller (line 56) can
handle/log/skip corrupted staging files.
---
Duplicate comments:
In `@src/parseable/staging/reader.rs`:
- Line 602: Rename the test function symbol
testget_reverse_reader_single_message to use snake_case
(test_get_reverse_reader_single_message) so it follows Rust naming conventions;
update the function declaration fn testget_reverse_reader_single_message() ->
io::Result<()> to fn test_get_reverse_reader_single_message() -> io::Result<()>
and update any references to that function within the file (e.g., test
registration or calls) to the new name.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 28749a8e-693e-4b15-8cee-518dc3fafbda
📒 Files selected for processing (2)
src/event/format/json.rssrc/parseable/staging/reader.rs
✅ Files skipped from review due to trivial changes (1)
- src/event/format/json.rs
parmesant
force-pushed
the
ingestion-traces
branch
from
04293d0 to
a3ea90e
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/handlers/http/modal/utils/ingest_utils.rs (1)
227-253:⚠️ Potential issue | 🟠 MajorAvoid side-effectful
event.process()insiderayon::try_for_each.Parallel execution with early return on error causes non-deterministic partial ingestion. When a worker fails, other workers may have already persisted their records. This creates inconsistent state where some records in the batch are ingested while others are not, making client retries unpredictable and risky.
Suggested direction
- let r = data.into_par_iter().try_for_each(|json| { + let events: Result<Vec<_>, _> = data + .into_par_iter() + .map(|json| { let origin_size = json_byte_size(&json); - - match (json::Event { json, p_timestamp }).into_event( - stream_name.to_owned(), - origin_size, - &schema, - static_schema_flag, - custom_partition.as_ref(), - time_partition.as_ref(), - schema_version, - StreamType::UserDefined, - p_custom_fields, - telemetry_type, - tenant_id, - ) { - Ok(event) => match event.process() { - Ok(_) => ControlFlow::Continue(()), - Err(e) => ControlFlow::Break(e.into()), - }, - Err(e) => ControlFlow::Break(e), - } - }); - if let ControlFlow::Break(e) = r { - return Err(PostError::Invalid(e)); + (json::Event { json, p_timestamp }).into_event( + stream_name.to_owned(), + origin_size, + &schema, + static_schema_flag, + custom_partition.as_ref(), + time_partition.as_ref(), + schema_version, + StreamType::UserDefined, + p_custom_fields, + telemetry_type, + tenant_id, + ) + }) + .collect(); + for event in events.map_err(PostError::Invalid)? { + event.process()?; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/handlers/http/modal/utils/ingest_utils.rs` around lines 227 - 253, The current use of data.into_par_iter().try_for_each calls side-effectful event.process() inside parallel workers, causing non-deterministic partial ingestion; change the logic to first transform/validate all inputs in parallel using json::Event::into_event (collecting Results into a Vec or short-circuiting on error), and only after successful collection iterate serially over the resulting events to call event.process(); on any transformation error return Err(PostError::Invalid(...)) as before, and on any process() error return Err(PostError::Invalid(...)), ensuring no persistence side-effects happen during the parallel phase (look for data.into_par_iter().try_for_each, json::Event::into_event, event.process(), ControlFlow::Break, and PostError::Invalid).
♻️ Duplicate comments (1)
src/parseable/streams.rs (1)
688-691:⚠️ Potential issue | 🟠 MajorClean up
processing_*files when all readers are rejected.If
MergedReverseRecordReader::try_new(&arrow_files)yields no readers, this branch skips cleanup and leaves the moved group stranded inprocessing_*.🧹 Proposed fix
let record_reader = MergedReverseRecordReader::try_new(&arrow_files); if record_reader.readers.is_empty() { + self.cleanup_arrow_files_and_dir(&arrow_files, tenant_id); continue; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/streams.rs` around lines 688 - 691, When MergedReverseRecordReader::try_new(&arrow_files) returns a reader struct with an empty readers list, we must delete the moved group from the processing_* location before continuing; update the branch that checks if record_reader.readers.is_empty() to invoke the existing cleanup/removal routine (or add one) that deletes the processing_* files associated with the current arrow_files/group and any moved marker, then continue. Locate the logic around MergedReverseRecordReader::try_new and the variable arrow_files in the loop and call the same cleanup helper used elsewhere (or implement a remove_processing_group function) to ensure no stranded processing_* files remain.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/handlers/http/modal/mod.rs`:
- Around line 133-140: The startup log hardcodes "Query server" in the
tracing::warn! within ParseableServer::start; change the message to include the
actual node role instead—use the role/name property from the ParseableServer
instance (e.g., PARSEABLE.role or self.role / ParseableServer::role()) in the
formatted string so ingestors/indexers/prism nodes emit their correct role at
startup, and update the tracing::warn! call to interpolate that dynamic role
value along with the existing options fields.
In `@src/handlers/http/modal/utils/ingest_utils.rs`:
- Around line 94-137: The OTEL branches (LogSource::OtelLogs,
LogSource::OtelTraces, LogSource::OtelMetrics) currently forward the incoming
time_partition into push_logs; change each push_logs call so it always passes
None for the time_partition argument (i.e., enforce time_partition = None) while
keeping the rest of the arguments and use of
flatten_otel_logs/flatten_otel_traces/flatten_otel_metrics unchanged so OTEL
ingestion cannot carry a time partition.
- Around line 194-224: The fast-path for custom_partition.is_none() is dropping
caller-provided time_partition by setting Event.time_partition = None (and in
other places hard-coding parsed_timestamp to now); update the Event construction
in this branch to preserve the caller's p_timestamp and time_partition: set
parsed_timestamp = p_timestamp.naive_utc() (not Utc::now()) and set
time_partition = time_partition.clone() or time_partition.map(|t| t.clone()) so
if a time_partition was provided it is kept; leave other fields (rb,
origin_size, is_first_event, custom_partition_values) unchanged.
In `@src/parseable/streams.rs`:
- Around line 156-167: When acquiring self.writer.lock() in the ingest/flush
paths you must not call poisoned.into_inner(); instead treat a poisoned mutex as
a fatal error and fail-fast: log the poisoning (including self.stream_name) and
then propagate the failure by returning an Err from the current function (or
panic if the surrounding API expects process-level failure) rather than
recovering the inner guard. Update the lock handling at the shown
self.writer.lock() site and the similar occurrence around lines 545-553 to
remove into_inner() usage and return/propagate a poison error so corrupted
writer state cannot be reused.
In `@src/telemetry.rs`:
- Around line 61-63: The code sets the traces exporter protocol using only
OTEL_EXPORTER_OTLP_PROTOCOL and defaults to "http/json"; change the lookup to
first check OTEL_EXPORTER_OTLP_TRACES_PROTOCOL, fall back to
OTEL_EXPORTER_OTLP_PROTOCOL if not set, and use "http/protobuf" as the final
default; update the variable named protocol (where std::env::var is called) to
implement this precedence and default so traces-specific env var overrides the
generic one and the default matches the OpenTelemetry spec.
---
Outside diff comments:
In `@src/handlers/http/modal/utils/ingest_utils.rs`:
- Around line 227-253: The current use of data.into_par_iter().try_for_each
calls side-effectful event.process() inside parallel workers, causing
non-deterministic partial ingestion; change the logic to first
transform/validate all inputs in parallel using json::Event::into_event
(collecting Results into a Vec or short-circuiting on error), and only after
successful collection iterate serially over the resulting events to call
event.process(); on any transformation error return Err(PostError::Invalid(...))
as before, and on any process() error return Err(PostError::Invalid(...)),
ensuring no persistence side-effects happen during the parallel phase (look for
data.into_par_iter().try_for_each, json::Event::into_event, event.process(),
ControlFlow::Break, and PostError::Invalid).
---
Duplicate comments:
In `@src/parseable/streams.rs`:
- Around line 688-691: When MergedReverseRecordReader::try_new(&arrow_files)
returns a reader struct with an empty readers list, we must delete the moved
group from the processing_* location before continuing; update the branch that
checks if record_reader.readers.is_empty() to invoke the existing
cleanup/removal routine (or add one) that deletes the processing_* files
associated with the current arrow_files/group and any moved marker, then
continue. Locate the logic around MergedReverseRecordReader::try_new and the
variable arrow_files in the loop and call the same cleanup helper used elsewhere
(or implement a remove_processing_group function) to ensure no stranded
processing_* files remain.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 0c5a0fe7-ccf1-41b3-9239-6a518f60a518
📒 Files selected for processing (16)
Cargo.tomlsrc/cli.rssrc/event/format/json.rssrc/event/format/mod.rssrc/event/mod.rssrc/handlers/http/modal/mod.rssrc/handlers/http/modal/utils/ingest_utils.rssrc/lib.rssrc/otel/metrics.rssrc/parseable/staging/reader.rssrc/parseable/streams.rssrc/storage/field_stats.rssrc/storage/object_storage.rssrc/sync.rssrc/telemetry.rssrc/utils/json/mod.rs
✅ Files skipped from review due to trivial changes (3)
- src/utils/json/mod.rs
- src/lib.rs
- src/storage/object_storage.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- src/storage/field_stats.rs
- src/sync.rs
parmesant
force-pushed
the
ingestion-traces
branch
from
a3ea90e to
b4d42ac
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (4)
src/parseable/streams.rs (2)
549-557:⚠️ Potential issue | 🟠 MajorDon’t recover a poisoned writer during flush.
push()now aborts on poison, butflush()still callsinto_inner()and keeps converting with potentially corrupted writer state. This path should fail fast as well.🔒 Safer handling
- let mut writer = match self.writer.lock() { - Ok(guard) => guard, - Err(poisoned) => { - error!( - "Writer lock poisoned while flushing data for stream {}", - self.stream_name - ); - poisoned.into_inner() - } - }; + let mut writer = self.writer.lock().expect(LOCK_EXPECT);Based on learnings, Parseable treats staging/invariant failures as critical and prefers fail-fast behavior over continuing in a potentially inconsistent state.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/streams.rs` around lines 549 - 557, The flush() path should fail fast instead of calling poisoned.into_inner(); in the match on self.writer.lock() inside flush(), replace the Err(poisoned) branch so it does not call into_inner() but instead logs the poison and returns/propagates an error (or panics) consistent with push()’s behavior; specifically, update the match in the flush() method (the block currently logging "Writer lock poisoned while flushing data for stream {}") to immediately abort/return an Err indicating a poisoned lock (do not call into_inner()) so the code treats a poisoned writer as a fatal invariant violation like push().
684-695:⚠️ Potential issue | 🟠 MajorDon’t strand moved Arrow groups under
processing_*.If
MergedReverseRecordReader::try_new()filters every file out, this branch justcontinues. Those files were already moved intoprocessing_*, and normal cycles only regroup the current in-process directory, so they fall off the retry path.Based on learnings, failed staged work is supposed to stay on the normal sync retry path; leaving it under an orphaned
processing_*directory bypasses that path.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/streams.rs` around lines 684 - 695, The loop calls MergedReverseRecordReader::try_new(...) and if record_reader.readers.is_empty() it currently continues, but those files were already moved into a processing_* dir and will be orphaned; instead, when record_reader.readers.is_empty() detect that case and move the corresponding arrow_files (or the parquet_path group) back from the processing_* location into the staged location (or call the existing undo/reenqueue helper) so they stay on the normal sync retry path, and update/reset metrics via update_staging_metrics/reset_staging_metrics as appropriate before continuing; specifically modify the branch around record_reader.readers.is_empty() to perform the requeue/rollback of the moved files rather than silently continue.src/telemetry.rs (1)
61-63:⚠️ Potential issue | 🟠 MajorHonor the traces-specific protocol override here.
This still ignores
OTEL_EXPORTER_OTLP_TRACES_PROTOCOL, so a traces-only config cannot override the generic protocol. The OTLP default is also expected to behttp/protobuf, nothttp/json.🔧 Suggested change
const OTEL_EXPORTER_OTLP_ENDPOINT: &str = "OTEL_EXPORTER_OTLP_ENDPOINT"; const OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: &str = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"; const OTEL_EXPORTER_OTLP_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_PROTOCOL"; +const OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL"; ... - let protocol = - std::env::var(OTEL_EXPORTER_OTLP_PROTOCOL).unwrap_or_else(|_| "http/json".to_string()); + let protocol = std::env::var(OTEL_EXPORTER_OTLP_TRACES_PROTOCOL) + .or_else(|_| std::env::var(OTEL_EXPORTER_OTLP_PROTOCOL)) + .unwrap_or_else(|_| "http/protobuf".to_string());OpenTelemetry OTLP exporter spec: does `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` override `OTEL_EXPORTER_OTLP_PROTOCOL` for traces, and what protocol should SDKs default to when neither is set?🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/telemetry.rs` around lines 61 - 63, The code sets protocol by only reading OTEL_EXPORTER_OTLP_PROTOCOL and defaulting to "http/json"; change the lookup in telemetry.rs so the traces-specific env var OTEL_EXPORTER_OTLP_TRACES_PROTOCOL is consulted first, falling back to OTEL_EXPORTER_OTLP_PROTOCOL if unset, and use the OTLP default "http/protobuf" as the final fallback; update the logic that assigns the variable protocol to reflect this precedence and default.src/handlers/http/modal/mod.rs (1)
133-140:⚠️ Potential issue | 🟡 MinorStartup log still hard-codes the node role as "Query server".
This is the shared startup path; please log the actual runtime role instead of a fixed "Query".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/handlers/http/modal/mod.rs` around lines 133 - 140, The startup log hardcodes "Query server"; change the tracing::warn format string to include the actual runtime role and pass the role value from the parsed config (use the runtime role field on PARSEABLE, e.g., PARSEABLE.role or PARSEABLE.options.role as appropriate). Update the message to something like "{} server with-\nNum workers: ..." and add the role value as the first argument so the logged role reflects the real runtime role instead of always "Query".
🧹 Nitpick comments (1)
src/parseable/staging/mod.rs (1)
41-42: SimplifyPoisonErrorvariant to plain string.
Mutex::lock()andRwLock::write()returnPoisonError<MutexGuard<_>>andPoisonError<RwLockWriteGuard<_>>respectively. The#[from] PoisonError<String>attribute will not auto-convert these—callers must manually constructPoisonError::new(String)anyway. Use a plainPoisonedLock(String)variant instead to make this synthesized error clearer and reduce unnecessary fabrication.♻️ Suggested change
- #[error("{0}")] - PoisonError(#[from] PoisonError<String>), + #[error("{0}")] + PoisonedLock(String),- return Err(StagingError::PoisonError(PoisonError::new(format!( + return Err(StagingError::PoisonedLock(format!( "Writer lock poisoned while ingesting data for stream {} - {}", self.stream_name, poisoned - )))); + )));🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/staging/mod.rs` around lines 41 - 42, Replace the PoisonError(#[from] PoisonError<String>) enum variant with a simple string-carrying variant (e.g., PoisonedLock(String)) and update its Display attribute (the existing #[error("{0}")]) accordingly; remove the #[from] on PoisonError since PoisonError<T> from parking_lot/std won't auto-convert, and update any call sites that constructed PoisonError::new(...) to instead produce the new PoisonedLock(String) variant (or convert the incoming PoisonError<T> into a meaningful String and wrap it) so callers no longer need to fabricate PoisonError<String>.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/handlers/http/modal/mod.rs`:
- Around line 24-25: The code currently casts the u64 worker count using
as_usize() (e.g., where num_workers.as_usize() is passed to
HttpServer::workers()), which can truncate on 32-bit targets; change this to use
usize::try_from(num_workers) and handle the Result explicitly: on Ok(n) pass n
to HttpServer::workers(), on Err(_) return or propagate a descriptive error (or
clamp/validate) so you never call HttpServer::workers() with a truncated or zero
value; update any function (e.g., the modal handler initialization) to propagate
the conversion error instead of relying on as_usize().
---
Duplicate comments:
In `@src/handlers/http/modal/mod.rs`:
- Around line 133-140: The startup log hardcodes "Query server"; change the
tracing::warn format string to include the actual runtime role and pass the role
value from the parsed config (use the runtime role field on PARSEABLE, e.g.,
PARSEABLE.role or PARSEABLE.options.role as appropriate). Update the message to
something like "{} server with-\nNum workers: ..." and add the role value as the
first argument so the logged role reflects the real runtime role instead of
always "Query".
In `@src/parseable/streams.rs`:
- Around line 549-557: The flush() path should fail fast instead of calling
poisoned.into_inner(); in the match on self.writer.lock() inside flush(),
replace the Err(poisoned) branch so it does not call into_inner() but instead
logs the poison and returns/propagates an error (or panics) consistent with
push()’s behavior; specifically, update the match in the flush() method (the
block currently logging "Writer lock poisoned while flushing data for stream
{}") to immediately abort/return an Err indicating a poisoned lock (do not call
into_inner()) so the code treats a poisoned writer as a fatal invariant
violation like push().
- Around line 684-695: The loop calls MergedReverseRecordReader::try_new(...)
and if record_reader.readers.is_empty() it currently continues, but those files
were already moved into a processing_* dir and will be orphaned; instead, when
record_reader.readers.is_empty() detect that case and move the corresponding
arrow_files (or the parquet_path group) back from the processing_* location into
the staged location (or call the existing undo/reenqueue helper) so they stay on
the normal sync retry path, and update/reset metrics via
update_staging_metrics/reset_staging_metrics as appropriate before continuing;
specifically modify the branch around record_reader.readers.is_empty() to
perform the requeue/rollback of the moved files rather than silently continue.
In `@src/telemetry.rs`:
- Around line 61-63: The code sets protocol by only reading
OTEL_EXPORTER_OTLP_PROTOCOL and defaulting to "http/json"; change the lookup in
telemetry.rs so the traces-specific env var OTEL_EXPORTER_OTLP_TRACES_PROTOCOL
is consulted first, falling back to OTEL_EXPORTER_OTLP_PROTOCOL if unset, and
use the OTLP default "http/protobuf" as the final fallback; update the logic
that assigns the variable protocol to reflect this precedence and default.
---
Nitpick comments:
In `@src/parseable/staging/mod.rs`:
- Around line 41-42: Replace the PoisonError(#[from] PoisonError<String>) enum
variant with a simple string-carrying variant (e.g., PoisonedLock(String)) and
update its Display attribute (the existing #[error("{0}")]) accordingly; remove
the #[from] on PoisonError since PoisonError<T> from parking_lot/std won't
auto-convert, and update any call sites that constructed PoisonError::new(...)
to instead produce the new PoisonedLock(String) variant (or convert the incoming
PoisonError<T> into a meaningful String and wrap it) so callers no longer need
to fabricate PoisonError<String>.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 56ee729e-6b4d-4573-b069-eb2c126b4640
📒 Files selected for processing (9)
src/cli.rssrc/handlers/http/modal/mod.rssrc/handlers/http/modal/utils/ingest_utils.rssrc/otel/metrics.rssrc/parseable/staging/mod.rssrc/parseable/staging/reader.rssrc/parseable/streams.rssrc/sync.rssrc/telemetry.rs
✅ Files skipped from review due to trivial changes (1)
- src/otel/metrics.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- src/sync.rs
- src/handlers/http/modal/utils/ingest_utils.rs
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/parseable/streams.rs (1)
691-695:⚠️ Potential issue | 🟠 MajorEmpty reader groups leave orphaned
processing_*directories.When
MergedReverseRecordReader::try_newreturns a reader with no valid readers (all files filtered out), the code justcontinues without cleaning up the arrow files that were already moved to theprocessing_*directory. These orphaned directories accumulate on disk and are never revisited.🐛 Proposed fix
for (parquet_path, arrow_files) in staging_files { let record_reader = MergedReverseRecordReader::try_new(&arrow_files); if record_reader.readers.is_empty() { + // Clean up arrow files that couldn't be read to avoid orphaned processing_* dirs + self.cleanup_arrow_files_and_dir(&arrow_files, tenant_id); continue; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/streams.rs` around lines 691 - 695, When MergedReverseRecordReader::try_new(...) returns an object whose record_reader.readers.is_empty(), the code currently continues without cleaning up the arrow_files that were already moved into their processing_* directories; fix this by, before the continue, iterating the arrow_files collection and removing each associated processing directory (e.g., compute the processing_* path from each arrow_file entry and call std::fs::remove_dir_all or the existing cleanup helper) and log non-fatal failures; keep using the same symbols (MergedReverseRecordReader::try_new, record_reader.readers.is_empty(), arrow_files, staging_files) and ensure removal errors are caught/ignored so the loop proceeds.
♻️ Duplicate comments (2)
src/parseable/staging/reader.rs (1)
607-607:⚠️ Potential issue | 🟡 MinorTypo in test function name: missing underscore.
The function
testget_reverse_reader_single_messageis missing an underscore between "test" and "get". Standard Rust test naming uses snake_case:test_get_reverse_reader_single_message.✏️ Proposed fix
- fn testget_reverse_reader_single_message() -> io::Result<()> { + fn test_get_reverse_reader_single_message() -> io::Result<()> {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/staging/reader.rs` at line 607, Rename the test function testget_reverse_reader_single_message to use snake_case by inserting the missing underscore, i.e., rename the symbol testget_reverse_reader_single_message to test_get_reverse_reader_single_message wherever it is declared and referenced (the test function in reader.rs), and update any test invocations or references to match the new name.src/parseable/streams.rs (1)
543-574:⚠️ Potential issue | 🟠 MajorInconsistent poison handling:
flush()still usesinto_inner().The
push()method was updated to return an error on poisoned mutex (lines 160-170), butflush()still recovers viapoisoned.into_inner()(lines 551-557). This is inconsistent and can persist corrupted writer state. Based on learnings, Parseable treats staging/invariant failures as critical and prefers fail-fast behavior.Additionally, since
flush()returns(), consider either:
- Changing the return type to
Result<(), StagingError>and propagating the error, or- Using
expect(LOCK_EXPECT)to panic on poison (consistent with other lock usages in this file like line 536).🐛 Proposed fix (option 2 - panic on poison)
pub fn flush(&self, forced: bool) { let _span = info_span!("flush", stream_name = %self.stream_name, forced).entered(); // Swap out stale writers under the lock, drop them after releasing it. // DiskWriter::Drop does I/O (IPC finish + file rename) so dropping // outside the lock avoids blocking concurrent push() calls. let stale_writers = { - let mut writer = match self.writer.lock() { - Ok(guard) => guard, - Err(poisoned) => { - error!( - "Writer lock poisoned while flushing data for stream {}", - self.stream_name - ); - poisoned.into_inner() - } - }; + let mut writer = self.writer.lock().expect(LOCK_EXPECT); writer.mem.clear();🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/parseable/streams.rs` around lines 543 - 574, flush() currently recovers from a poisoned mutex via poisoned.into_inner(), which is inconsistent with push() and risks continuing with corrupted state; change the poison handling in flush() to be fail-fast like other usages by replacing the match on self.writer.lock() with self.writer.lock().expect(LOCK_EXPECT) (or alternatively change flush() signature to return Result<(), StagingError> and propagate the lock error), ensuring you reference the flush() function and the same LOCK_EXPECT constant used elsewhere so poison leads to a panic rather than silent recovery.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/handlers/http/modal/utils/ingest_utils.rs`:
- Around line 52-64: The tracing spans are currently using fields(stream_name)
which creates an empty field instead of capturing the parameter; update the
instrument attributes to explicitly pass the value (e.g., change
fields(stream_name) to fields(stream_name = %stream_name)) or remove stream_name
from fields(...) to rely on automatic recording; apply this fix to the
flatten_and_push_logs instrument attribute and the other instrument attribute
later in the file where fields(stream_name) is used so the actual stream_name
value is recorded.
---
Outside diff comments:
In `@src/parseable/streams.rs`:
- Around line 691-695: When MergedReverseRecordReader::try_new(...) returns an
object whose record_reader.readers.is_empty(), the code currently continues
without cleaning up the arrow_files that were already moved into their
processing_* directories; fix this by, before the continue, iterating the
arrow_files collection and removing each associated processing directory (e.g.,
compute the processing_* path from each arrow_file entry and call
std::fs::remove_dir_all or the existing cleanup helper) and log non-fatal
failures; keep using the same symbols (MergedReverseRecordReader::try_new,
record_reader.readers.is_empty(), arrow_files, staging_files) and ensure removal
errors are caught/ignored so the loop proceeds.
---
Duplicate comments:
In `@src/parseable/staging/reader.rs`:
- Line 607: Rename the test function testget_reverse_reader_single_message to
use snake_case by inserting the missing underscore, i.e., rename the symbol
testget_reverse_reader_single_message to test_get_reverse_reader_single_message
wherever it is declared and referenced (the test function in reader.rs), and
update any test invocations or references to match the new name.
In `@src/parseable/streams.rs`:
- Around line 543-574: flush() currently recovers from a poisoned mutex via
poisoned.into_inner(), which is inconsistent with push() and risks continuing
with corrupted state; change the poison handling in flush() to be fail-fast like
other usages by replacing the match on self.writer.lock() with
self.writer.lock().expect(LOCK_EXPECT) (or alternatively change flush()
signature to return Result<(), StagingError> and propagate the lock error),
ensuring you reference the flush() function and the same LOCK_EXPECT constant
used elsewhere so poison leads to a panic rather than silent recovery.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 75447c63-646e-48de-8b5b-44218a2075ae
📒 Files selected for processing (9)
src/cli.rssrc/handlers/http/modal/mod.rssrc/handlers/http/modal/utils/ingest_utils.rssrc/otel/metrics.rssrc/parseable/staging/mod.rssrc/parseable/staging/reader.rssrc/parseable/streams.rssrc/sync.rssrc/telemetry.rs
✅ Files skipped from review due to trivial changes (2)
- src/parseable/staging/mod.rs
- src/otel/metrics.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- src/handlers/http/modal/mod.rs
- src/telemetry.rs
parmesant
force-pushed
the
ingestion-traces
branch
from
b4d42ac to
845ccb2
Compare
parmesant
force-pushed
the
ingestion-traces
branch
from
845ccb2 to
229e486
Compare
parmesant
force-pushed
the
ingestion-traces
branch
from
229e486 to
1ea46fa
Compare
nikhilsinhaparseable
approved these changes
Apr 28, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #XXXX.
Description
This PR has:
Summary by CodeRabbit
New Features
Performance Improvements
Observability
Reliability
Breaking Changes