feat(clp-tdl-package): Implement the clp-s S3 compression task. - #2411
Conversation
The initial and maximum job-state poll backoffs were fixed constants inside the Spider implementation, so a caller had no way to tune how aggressively a job is polled. Both are now `Duration` parameters of `run_s3_compression_job_to_completion`, leaving the pacing decision with the caller that knows how long its jobs typically run.
# Conflicts: # Cargo.lock # Cargo.toml
…asks need: * Add `ArchiveOutputStorage` and the archive output's `storage` and `retention_period` fields. * Add `Database`'s fixed `table_prefix`. * Fill in `SpiderTaskExecutorConfig` and resolve its relative paths against `CLP_HOME`. * Add `resolve_dataset_name` and the default dataset name. * Add `generate_s3_url`.
…s stderr-join deadlock.
… S3 client into a helper.
|
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:
WalkthroughChangesAdds the CLP TDL compression
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant SpiderTask
participant compress
participant clp-s
participant ArchiveFinisher
participant S3
participant indexer
SpiderTask->>compress: invoke S3 compression task
compress->>clp-s: run compression
clp-s-->>compress: emit archive statistics
compress->>ArchiveFinisher: schedule archive finalization
ArchiveFinisher->>S3: upload archive
ArchiveFinisher->>indexer: index archive
ArchiveFinisher-->>compress: return completion result
compress-->>SpiderTask: return compression output
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
sitaowang1998
left a comment
sitaowang1998
left a comment
There was a problem hiding this comment.
I am not familiar with clp s3 details, so I assume that the overall algorithm is correct during my review.
| credential_env: Vec<(&'static str, String)>, | ||
| tmp_file_deleter: &mut TmpFileDeleter, | ||
| ) -> anyhow::Result<(ClpSInput, Vec<(&'static str, String)>)> { | ||
| if !clp_s_option.unstructured { |
There was a problem hiding this comment.
Not really in scope of this review, but we should avoid using boolean flag here. I understand that for serialization a boolean is needed, but that should be a serialization details, hidden from user-facing code.
There was a problem hiding this comment.
Hmmm, I'm not sure I fully understand. Can you elaborate on the desired interface?
There was a problem hiding this comment.
If should be an enum.
There was a problem hiding this comment.
Hmmm, I know this is out of the scope for this PR. But I'm not sure if we really need an enum for this. If we use a binary flag to determine whether the input source is from s3, for example, is_from_s3, which differs from "from filesystem", then I think it would be a good practice to define an enum for it, even it only has two options. However, if an option is purely a binary flag of two opposite concepts like "structured" and "unstructured", I don't really see the benefit of using an enum. You can even match this flag like match unstructured { true => xxx, false => xxx, } with the readability preserved. So I'd argue that using a binary flag for this case should be sufficient.
Added the scenarios I've tested in the PR description. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@components/clp-rust-utils/src/s3/url.rs`:
- Around line 1-27: Update generate_s3_url to percent-encode object_key before
composing either endpoint or AWS URLs, preserving each slash as a path separator
while encoding spaces, reserved characters, percent signs, and non-ASCII bytes.
Reuse an appropriate URL/path encoding utility and apply the encoded key in all
URL construction branches.
In `@components/clp-tdl-package/Cargo.toml`:
- Line 16: Update the spider-tdl dependency declaration to use a fixed reviewed
git revision via rev, or a specific published version, instead of branch =
"main"; preserve the existing derive feature and dependency source.
In `@components/clp-tdl-package/src/task/compression/mod.rs`:
- Around line 30-33: Update the compression::commit commit_task implementation
so it no longer unconditionally panics via unimplemented!. Before retaining its
registration, either implement the task, return an explicit TdlError execution
failure, or remove compression::commit from the package task list until it is
ready.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: d31a0634-1652-4a3b-a07d-4a1de125e030
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (14)
Cargo.tomlcomponents/clp-rust-utils/src/clp_config/package/config.rscomponents/clp-rust-utils/src/dataset.rscomponents/clp-rust-utils/src/s3.rscomponents/clp-rust-utils/src/s3/url.rscomponents/clp-rust-utils/src/task_io/compression.rscomponents/clp-tdl-package/Cargo.tomlcomponents/clp-tdl-package/src/common.rscomponents/clp-tdl-package/src/lib.rscomponents/clp-tdl-package/src/task/compression/commit.rscomponents/clp-tdl-package/src/task/compression/compress.rscomponents/clp-tdl-package/src/task/compression/mod.rscomponents/clp-tdl-package/src/task/mod.rstaskfile.yaml
| /// Generates the URL of an S3 object. | ||
| /// | ||
| /// When `endpoint_url` is set, a path-style URL rooted at that endpoint is produced and | ||
| /// `region_code` is ignored, since the endpoint already fully specifies the host. Otherwise, a | ||
| /// virtual-hosted-style URL against the default AWS endpoint is produced, incorporating | ||
| /// `region_code` when it is set. | ||
| /// | ||
| /// # Returns | ||
| /// | ||
| /// The generated object URL. | ||
| #[must_use] | ||
| pub fn generate_s3_url( | ||
| endpoint_url: Option<&str>, | ||
| region_code: Option<&str>, | ||
| bucket: &str, | ||
| object_key: &str, | ||
| ) -> String { | ||
| endpoint_url.map_or_else( | ||
| || { | ||
| region_code.map_or_else( | ||
| || format!("https://{bucket}.s3.amazonaws.com/{object_key}"), | ||
| |region| format!("https://{bucket}.s3.{region}.amazonaws.com/{object_key}"), | ||
| ) | ||
| }, | ||
| |endpoint| format!("{}/{bucket}/{object_key}", endpoint.trim_end_matches('/')), | ||
| ) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
clp-s clp-core --files-from S3 URL format object key encoding
💡 Result:
When working with CLP (Compressed Log Processor) tools like clp-s and the associated S3 compression scripts, the expected S3 URL formats are standard virtual-hosted or path-style URL structures [1]. S3 URL Formats: The tools accept S3 locations in either of these two formats [1]: 1. Virtual-hosted style: https://.s3..amazonaws.com/ 2. Path-style: https://s3..amazonaws.com// Using --files-from (or --inputs-from): The --inputs-from flag (often used in the sbin/compress-from-s3.sh script) accepts a path to a text file that contains one S3 object URL per line, adhering to the formats listed above [1][2]. Object Key Encoding: CLP commands that accept URLs, such as clp-s, treat these inputs as network resources [3][4]. When authentication is enabled (via --auth s3), the tools issue GET requests following the AWS Signature Version 4 specification [3][4]. While CLP itself handles the network requests, it generally expects the object keys provided in the URL to follow standard URL encoding if they contain special characters (such as spaces or non-ASCII characters) to ensure compatibility with HTTP transport protocols and AWS S3 API requirements. Authentication: When using URLs, ensure that the environment variables AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and (optionally) AWS_SESSION_TOKEN are properly set in your environment, as the tools rely on these for AWS Signature Version 4 authentication [3][4].
Citations:
- 1: https://docs.yscope.com/clp/main/user-docs/guides-using-object-storage/aws-s3/using-clp-with-aws-s3.html
- 2: https://docs.yscope.com/clp/v0.11.0/user-docs/guides-using-object-storage/aws-s3/using-clp-with-aws-s3.html
- 3: https://docs.yscope.com/clp/main/user-docs/core-clp-s.html
- 4: https://docs.yscope.com/clp/v0.5.0/user-guide/core-clp-s.html
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== url.rs ==\n'
sed -n '1,200p' components/clp-rust-utils/src/s3/url.rs
printf '\n== compress.rs references ==\n'
rg -n "generate_s3_url|files-from|inputs-from|s3 url|S3" components/clp-tdl-package/src/task/compression/compress.rs components/clp-rust-utils -g '!target'
printf '\n== tests referencing generate_s3_url ==\n'
rg -n "generate_s3_url|s3.amazonaws.com|endpoint_url|region_code" components -g '!target'Repository: y-scope/clp
Length of output: 30689
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== Rust compression flow around build_s3_logs_list ==\n'
sed -n '300,340p' components/clp-tdl-package/src/task/compression/compress.rs
printf '\n== Python S3 URL generation/parsing ==\n'
sed -n '257,290p' components/clp-py-utils/clp_py_utils/s3_utils.py
sed -n '220,255p' components/clp-py-utils/clp_py_utils/s3_utils.py
printf '\n== Any explicit URL parsing/decoding in Rust S3 code ==\n'
rg -n "parse_s3_url|url::Url|percent|decode|encode|object_key" components/clp-rust-utils/src components/clp-tdl-package/src -g '!target'Repository: y-scope/clp
Length of output: 6872
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== S3InputSource definition ==\n'
sed -n '1,120p' components/clp-rust-utils/src/task_io/compression.rs
printf '\n== build_s3_logs_list tests and fixture data ==\n'
sed -n '820,880p' components/clp-tdl-package/src/task/compression/compress.rs
printf '\n== any object key validation/sanitization ==\n'
rg -n "object_keys|key_prefix|sanitize|encode|percent|urlencoding|Url::parse|parse\(" components/clp-rust-utils/src components/clp-tdl-package/src components/clp-py-utils/clp_py_utils -g '!target'Repository: y-scope/clp
Length of output: 11315
Percent-encode S3 object keys before composing the URL. object_key is inserted raw into the path, so valid keys containing spaces, #, ?, %, or non-ASCII characters can turn into malformed or ambiguous URLs in the --files-from list. Preserve / as a path separator while encoding the rest.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@components/clp-rust-utils/src/s3/url.rs` around lines 1 - 27, Update
generate_s3_url to percent-encode object_key before composing either endpoint or
AWS URLs, preserving each slash as a path separator while encoding spaces,
reserved characters, percent signs, and non-ASCII bytes. Reuse an appropriate
URL/path encoding utility and apply the encoded key in all URL construction
branches.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
components/clp-tdl-package/src/task/compression/compress.rs (4)
884-904: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for empty S3 object keys.
This change introduces an error path for empty keys, but the test exercises only valid keys. Add a case with
object_keys: vec![String::new()]and assert thatbuild_s3_logs_listfails.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@components/clp-tdl-package/src/task/compression/compress.rs` around lines 884 - 904, Extend the build_s3_logs_list_default_endpoint test coverage with an input_source whose object_keys contains an empty String, then assert that build_s3_logs_list returns an error for that case.
80-89: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winRegister the temporary list before writing it.
If
std::fs::writecreates a partial file and then fails, the?returns beforelist_pathis added toTmpFileDeleter, leaving temporary data behind. Build the list first, register the path, then write it; also treat a missing path as a normal cleanup outcome.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@components/clp-tdl-package/src/task/compression/compress.rs` around lines 80 - 89, Update the temporary S3 logs list flow around build_s3_logs_list, list_path, and TmpFileDeleter: build the list contents first, register list_path before calling std::fs::write, and make cleanup treat a missing path as a normal outcome. Preserve the existing write context and error logging while ensuring partial files are removed when writing fails.
121-138: 🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy liftBound concurrent archive finalisation and stream uploads.
Every archive immediately spawns a finisher, while
upload_file_to_s3(Line [588-615]) reads the complete archive into memory. A large partition can therefore queue unbounded work and retain multiple archive-sized buffers, causing severe memory pressure or OOM. Use a semaphore/bounded queue and stream or otherwise cap upload buffering.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@components/clp-tdl-package/src/task/compression/compress.rs` around lines 121 - 138, Bound archive finalization concurrency in the archive_callback flow using a semaphore or bounded queue, and ensure upload_file_to_s3 does not retain unbounded archive-sized buffers by streaming uploads or applying an equivalent buffer cap. Preserve archive processing while limiting the number of concurrent finishers and stream uploads.
262-284: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftPrevent inconsistent S3 and index state.
tokio::join!publishes upload and indexing concurrently and only checks their results after both have finished. Ifrun_indexercommits indexing state whileupload_file_to_s3later fails, this job reports failure despite visible search metadata for an archive without the archive object; the opposite failure can leave an orphaned S3 object. Make publishing atomic by ordering it (index after upload, or upload after index) or adding compensation/reconciliation that rolls back either side.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@components/clp-tdl-package/src/task/compression/compress.rs` around lines 262 - 284, Replace the concurrent tokio::join! flow around upload_file_to_s3 and index with an ordered or compensating publication flow so S3 and indexing cannot remain inconsistent. Ensure indexing occurs only after a successful upload, or roll back the upload if indexing fails; preserve the existing error context and logging for both operations.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@components/clp-tdl-package/src/task/compression/compress.rs`:
- Around line 884-904: Extend the build_s3_logs_list_default_endpoint test
coverage with an input_source whose object_keys contains an empty String, then
assert that build_s3_logs_list returns an error for that case.
- Around line 80-89: Update the temporary S3 logs list flow around
build_s3_logs_list, list_path, and TmpFileDeleter: build the list contents
first, register list_path before calling std::fs::write, and make cleanup treat
a missing path as a normal outcome. Preserve the existing write context and
error logging while ensuring partial files are removed when writing fails.
- Around line 121-138: Bound archive finalization concurrency in the
archive_callback flow using a semaphore or bounded queue, and ensure
upload_file_to_s3 does not retain unbounded archive-sized buffers by streaming
uploads or applying an equivalent buffer cap. Preserve archive processing while
limiting the number of concurrent finishers and stream uploads.
- Around line 262-284: Replace the concurrent tokio::join! flow around
upload_file_to_s3 and index with an ordered or compensating publication flow so
S3 and indexing cannot remain inconsistent. Ensure indexing occurs only after a
successful upload, or roll back the upload if indexing fails; preserve the
existing error context and logging for both operations.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: a561bddb-9791-412f-8172-72722c3537f3
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (1)
components/clp-tdl-package/src/task/compression/compress.rs
2 participants
Description
This PR depends on #2406.
This PR fills in the compression task registered by
clp-tdl-package: it compresses one partition of S3 objects into single-file archives withclp-s, indexes each archive, and uploads it to S3, returning the archives' metadata for the commit task. This is the worker behind thecompression::clp_s_s3_compressTDL task; the commit task remains unimplemented.Compression flow
compressmirrors the Python compression task:--files-fromlist file in the tmp directory.clp-swith--print-archive-stats, streaming each archive's stats line off stdout as it is produced.put_object) and runs theindexerover it, then the staged copy is cleaned up.Archives are streamed rather than collected up front so uploads and indexing overlap
clp-sstill running, and the finishers are drained before the task returns so no work is abandoned. The first finisher error (or aclp-sfailure) fails the task.Unstructured (log-converter) input
Structured (JSON) input is compressed directly from S3. Unstructured text input is first converted by
log-converterinto a local directory thatclp-sthen compresses, selected by theunstructuredflag on the task'sclp-soptions. Following the Python task, the unstructured path drops S3 auth from theclp-sinvocation (it reads the local converted files, not S3) and pins the timestamp key totimestamp; onlylog-convertercarries the S3 credentials in that case.Subprocess I/O
Both child processes capture stderr so it can be surfaced on failure.
clp-spipes both stdout and stderr, so its stderr is drained on a dedicated thread to avoid a pipe-buffer deadlock while stdout is streamed, and it is killed and reaped before an error is returned.log-converterdiscards stdout (/dev/null) and has only a single stderr pipe, so a straight read-to-EOF-then-wait is deadlock-free there; the wait is issued unconditionally so the child is always reaped even if reading its stderr fails.Temp-file cleanup
TmpFileDeleterregisters the list file, the converted directory, and each archive's staging path, and deletes them all whencompressreturns — on success or on any early error. Deletion is sequential and blocking (a handful of local paths,unlinkhas no async form, and a destructor cannot be async), and a path that cannot be deleted is logged as a warning rather than propagated.Observability
Each stage logs structured
tracingevents; every subprocess or S3 failure is logged with its captured context (exit status, stderr, archive id, S3 key) at the point it occurs.Checklist
breaking change.
Validation performed
compressend-to-end (outside Spider) againstclp-s,indexer, andlog-converterfrombuild/core, a local MariaDB, and a real S3 bucket, with a JSONtracingsubscriber capturing logs. Each scenario asserts the return value, the S3 side effect, and that the tmp directory is left clean:unstructured: falseOkwith the produced archive's metadata; the archive object is present in S3 under<key_prefix>/<dataset>/<archive_id>; the dataset'scolumn_metadatatable gains rows (indexer ran)started+completed successfullyinfo linesunstructured: trueOkwith archive metadata; archive object present in S3 (log-converter ran before clp-s, which compressed the local converted directory)started+completed successfullyinfo lines-converteddirectory removedErr(indexer exits non-zero)errorlineindexer exited on failure.withstatus+archive_pathobject_keysreferencing a non-existent S3 keyErr(clp-s exits non-zero)errorlineclp-s exited on failure.with the captured clp-s stderr (S3 read, 404)Err(clp-s exits non-zero on the credential path forwarded to the child)errorlineclp-s exited on failure.with the captured clp-s stderr (S3 read, 403)Err(clp-s exits non-zero)errorlineclp-s exited on failure.with the captured clp-s stderr (JSON parse error, after the object was read)Note: surfaced by the DB-unreachable and malformed-content cases: because a finisher uploads to S3 and runs the indexer concurrently, once an archive is produced its S3 upload is not rolled back if the indexer (or the overall task) then fails. A failed task can thus leave an orphan archive object in S3 (local staging is still cleaned up); the object has no metadata row, so it is invisible to the commit task and left to retention/GC. This is a known behavior and will be addressed in the future.
Summary by CodeRabbit