[gerashegalov]

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.

[greptile-apps]

Confidence Score: 5/5

[Medium risk] Adds incremental resource accounting with local caching and timeouts.

The PR appears safe to merge, though previously populated event-log caches will be downloaded again.

Findings

1.  Existing event caches are bypassed ▶

Summary

The PR adds application-scoped, incremental ResourceManager accounting with a persistent SQLite index, resumable Spark event-log processing, and an optional time budget.

• The latest change restores guidance for diagnosing permanent cache failures and safely rebuilding the index.
• The earlier finding that existing event-objects-v1/ caches are bypassed remains outstanding; it is not reposted here.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[RM object listing] --> B[Refresh changed objects in SQLite index]
  B --> C[Select application and cluster evidence]
  C --> D[Read or resume Spark event metadata]
  D --> E[Calculate application usage]

Loading

_{Reviews (9) · Last reviewed commit: "Document recovery from permanent YARN ca..."}

[tgravescs]

this needs comments

[gerashegalov]

Addressed in efa400e: the download helper now documents why indexed/event-log reads require the listed version while legacy YARN reads do not, and an inline comment explains the event-object cache identity, completeness check, and atomic publication.

[tgravescs]

it would be nice to have a useful name instead of blob

[gerashegalov]

Addressed in efa400e: renamed blob to cached_object and blob_dir to object_cache_dir.

[tgravescs]

It think renaming this to be check timeout or something would make it more clear.. budget could be file system, time, etc..

[gerashegalov]

Addressed in efa400e: renamed the local callback to check_timeout and the surrounding value to timeout_seconds.

[tgravescs]

seems it only changed some of the places?

[gerashegalov]

Good catch. I renamed check_budget to check_timeout throughout.

[tgravescs]

Medium: the default path can stay pending forever (yarn-resource-cost/yarn_job_cost_api.py:182)

_download_objects now always sends a conditional GET (IfMatch=). It treats PreconditionFailed, NoSuchKey and a length mismatch as ArchivePendingError, which means "retry later". That also applies to the default rm_only=False path.

• What goes wrong: that path lists the whole cluster LogUri, then downloads every RM and NodeManager log one at a time. If the active hadoop-yarn-resourcemanager*.log is rewritten faster than the listing plus the earlier downloads take, its ETag changes before its GET. The GET fails with 412 and the request returns pending.
• Why it never recovers: this path keeps no index or cache between attempts, so every retry hits the same race. Before this change, it just downloaded the newer object and finished.
• Why rm_only is fine: that path re-lists right before indexing to keep the gap between listing and download short. The default path doesn't.

Possible fixes:

• Only send IfMatch on the rm_only/indexed path, which needs a consistent snapshot.
• Or, on the default path, re-list and retry just the object that changed instead of failing the whole request.

Checked and found sound:

• The index generation fence ordering.
• Per-object SQLite transactions and rollback.
• SQLite retry classification on Python 3.10 and 3.11+.
• Checkpoint chaining on a cache hit.
• The fallback for tar member stream identity.
• Publishing files into the event-object cache.

Smaller possible issues I noticed earlier (not confirmed):

• In _download_objects, symlink_to fails if two keys turn into the same local path.
• With caching off, a failed download leaves a partial file at target.

[gerashegalov]

Fixed in efa400e and finalized in 2090cce. Legacy YARN archive GETs no longer pin to an earlier listing ETag. They validate bytes against the GET response's ContentLength, so active-log replacements can complete while short reads cannot be counted. Clients omitting that boto3 response field now fail clearly. RM-only/indexed and event-log reads retain listed-version checks. Local suite: 119 tests, 5 subtests passed. The two smaller speculative concerns remain non-actionable without a reproducer: uncached partial files are discarded with the per-call TemporaryDirectory, and normal EMR event-log keys do not collapse to the same symlink path.

[tgravescs]

this to me is to much detail. I'd like a human readable summary of functionality, how to use, dependencies, etc. I don't need a line by line explanation or detailed design doc.

more like using a cache that is backed by sqllite, files stored in X/Y.. requires Z,

[gerashegalov]

Agreed—the section was too implementation-heavy. In 7d90950 I replaced it with a shorter explanation of RM-only mode, an example showing rm_only=True with a persistent cache_dir, dependencies and cache locations, and the operational points callers need for retries and timeouts.

[tgravescs]

I assumed index rm and event-log reads here mean cached objects. this should be made high level description of the fuction. indexed rm implies I already know about caching.

I don't know what "scan the whole cluster without checkpoints" means

[gerashegalov]

Agreed. I shortened the docstring to focus on what the downloader does.

[tgravescs]

why is this named "event-objects-v1"? this is download objects function, this doesn't have to be event logs, correct?

[gerashegalov]

Correct: the downloader is generic, although currently only S3 event-log calls pass a persistent cache_dir. In 7d90950 I renamed the directory to s3-objects-v1 and updated the documentation and tests. There is no production cache to migrate; existing development caches under the old name will simply be cold once.

[tgravescs]

I think the growth here of the sql database and files still grows forever. Well as many a logs in s3. We could prune all old entires on RM reboot for sure. it seems like we could also add just looking back for a certain period of time instead of til the end of time. This can be a followup but something to keep in mind, especially for customer clusters

The 120 seconds timeout could happen with very large event logs. What is the real intent of the timeout check?

[greptile-apps]

Existing event caches are bypassed

If a caller reuses a cache populated before this change, the new s3-objects-v1/ lookup cannot find event logs stored in event-objects-v1/. Those logs are downloaded again, their file-identity-based parsing checkpoints cannot be reused, and the old copies remain on disk. For large logs, this can delay a bounded accounting attempt and duplicate substantial storage.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

[tgravescs]

the event log cache is only useful if it started download timed out and then retried, correct? otherwise we cache event logs that should never be used again. Am I understanding that properly?

[gerashegalov]

the event log cache is only useful if it started download timed out and then retried, correct? otherwise we cache event logs that should never be used again. Am I understanding that properly?

That's correct. If all segments of the event log were downloaded successfully without timeout the cached segments are not reused once the cost calculation is done. We could add code to drop them from cache upon completion of the app cost request.