Repository navigation
Change to allow passing s3 uri to pick the logs from a separate s3 directory and limit to just resource manager logs - #277
Merged
Conversation
Signed-off-by: Thomas Graves <tgraves@nvidia.com>
Contributor
 |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Zero-byte ResourceManager logs from the log shipper no longer count as downloaded, so both the API and the CLI fall back to the EMR log URI. Refreshing a log cache now removes the previous snapshot so logs rotated or deleted upstream are not kept. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Thomas Graves <tgraves@nvidia.com>
Refresh now only removes the .download-complete marker, so drop the stale-file assertion. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
gerashegalov
left a comment
gerashegalov
left a comment
Collaborator
There was a problem hiding this comment.
I agree with the reproducibility concern #277 (comment)
Shipper source missing from JSON When shipped logs supply an --event-log-root run, this branch clears emr_log_uri, but the JSON output does not record --yarn-log-shipper-uri instead. The result has no S3 source for its YARN evidence, making the cost calculation harder to trace or reproduce. Signed-off-by: Thomas Graves <tgraves@nvidia.com>
gerashegalov
self-requested a review
September 30, 2026 20:22
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
materialize_yarn_logs now returns the S3 URI it actually read from (log shipper or EMR LogUri) alongside the local path, so --output-json can record it. Add a test for source selection and assert the field in the end-to-end JSON test. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A local shipper path skipped the empty-log check and never fell back to the EMR log archive. Match the API, which already requires an S3 URI. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Comment on lines
+2021
to
+2024
| if args.yarn_log_shipper_uri is not None and not args.yarn_log_shipper_uri.startswith( | ||
| ("s3://", "s3a://", "s3n://") | ||
| ): | ||
| parser.error("--yarn-log-shipper-uri must be an S3 URI") |
Contributor
There was a problem hiding this comment.
s3:///logs passes this scheme-only check, even though the API rejects it as an invalid S3 URI. The CLI then passes it to aws s3 cp, so a typo fails during log download rather than being rejected as an invalid --yarn-log-shipper-uri argument.
Collaborator
There was a problem hiding this comment.
OK, to defer to a follow up
gerashegalov
added a commit
that referenced
this pull request
Oct 5, 2026
## Summary Add opt-in, application-scoped incremental ResourceManager accounting for long-running EMR clusters. Repeated attempts currently re-download and parse historical cluster logs; this change retains completed work and avoids NodeManager fallback in the low-latency mode. - Add `rm_only=True`, a persistent `cache_dir`, and a cooperative `timeout_seconds` budget to the Python API. A configured log-shipper prefix is authoritative: missing RM evidence returns retryable pending without scanning the full EMR archive or reading Spark event logs prematurely. The existing diagnostic/default path remains available. - Cache normalized RM records in a per-source SQLite index, refreshing only new or changed objects and selecting the requested application's records plus historical node/scheduler metadata. Use conditional GETs, atomic per-object updates, deletion reconciliation, and a final snapshot check to prevent partial or stale evidence from becoming complete usage. - Preserve complete Spark event-log downloads and per-segment parsing checkpoints across attempts. Invalidate successor checkpoints on source changes, prioritize cold RM objects, and let event work progress after the target RM summary has been observed. After prioritized event work, list a fresh full RM snapshot rather than reusing stale ETags from the initial availability probe. - Add phase timing and object/byte/cache-reuse diagnostics, regression coverage, packaging support, and operating guidance. - Fence RM listings against concurrent index writes, reject event downloads whose byte count differs from the listed object, and preserve unbounded processing for the original API mode when no timeout is requested. - Treat SQLite lock contention and interrupted queries as retryable; propagate permanent cache/storage failures instead of reporting them as missing logs. Python 3.10 is covered by a narrow fallback because its SQLite exceptions lack error codes. Builds on the log-shipper API in #277. Resource-seconds and accounting completeness semantics are unchanged; the caller still owns pricing and retry scheduling. ## Validation - `python -m pytest -q yarn-resource-cost`: **119 passed, 5 subtests passed** locally on Python 3.11; PR CI runs Python 3.10, 3.11, and 3.12. - `python -m py_compile yarn-resource-cost/*.py`: passed. - `package_yarn_job_cost.py` standalone bundle build and verification: passed on the latest change. - `git diff --check`: passed on the latest change. - Independent local review exercised conditional-GET replacement/deletion races, concurrent index refreshes, multi-page listings, checkpoint invalidation, and bounded-retry progress. A deterministic periodic-upload reproduction completes on attempt two in **17 simulated seconds**, with zero conditional GET failures. This is a regression check, not a production benchmark. - Review regressions reproduce a stale RM listing after a newer index write, a silent short event-log read, an oversized legacy archive, and a malformed persistent SQLite cache. Tests verify that only transient conditions return pending and that permanent cache errors propagate. ## Operational limits In RM-only mode, the omitted budget defaults to 120 seconds; the original mode remains unbounded unless a timeout is explicitly supplied. The budget is cooperative, not a hard deadline; injected SDK clients need bounded I/O settings. Each object download or segment parse must fit an attempt, and the final mutable RM snapshot plus warm-cache overhead must also fit. Cold calls still scan RM history; warm calls still list the prefix and process global metadata. The disposable cache requires private local storage with SQLite locking and has no automatic TTL or size eviction; it also retains full Spark event-log objects. Permanent cache errors require operator attention or a stopped-caller cache rebuild. Live AWS and representative production-scale cold/warm latency have not been measured. --------- Signed-off-by: Gera Shegalov <gshegalov@nvidia.com>
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
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.
The emr log pusher defaults to push logs to s3 at a 5 minutes interval. It isn't configurable. Some people want this more frequent and could push logs with fluent bit or other utility. Allow reading from another s3 directory and only looking at resource manager logs.
calculate_yarn_job_cost.py has not been manually tested by me.