Incremental RM-only collection for YARN Cost Tool

.github/workflows/yarn-resource-cost-python.yml

@@ -28,9 +28,9 @@ jobs:
       - name: Compile
         run: python -m py_compile yarn-resource-cost/*.py
       - name: Install
-        run: python -m pip install ./yarn-resource-cost
+        run: python -m pip install ./yarn-resource-cost pytest
       - name: Unit tests
-        run: python -m unittest discover -s yarn-resource-cost -p 'test*.py'
+        run: python -m pytest -q yarn-resource-cost
       - name: Installed console error handling
         run: |
           python - <<'PY'

yarn-resource-cost/CONTRIBUTING.md

@@ -10,8 +10,8 @@ explicitly sanitized fixtures.
 Before submitting a change, run:
 
 ```bash
-python3 -m pip install ./yarn-resource-cost
+python3 -m pip install ./yarn-resource-cost pytest
 python3 -m py_compile yarn-resource-cost/*.py
-python3 -m unittest discover -s yarn-resource-cost -p 'test*.py'
+python3 -m pytest -q yarn-resource-cost
 python3 yarn-resource-cost/package_yarn_job_cost.py
 ```

yarn-resource-cost/README.md

@@ -80,6 +80,78 @@ when a log shipper uploads them sooner. Only file names containing
 the `LogUri` archive is used instead. The `calculate_yarn_job_cost.py` CLI
 accepts the same prefix as `--yarn-log-shipper-uri`.
 
+### Low-latency collection on persistent clusters
+
+Set `rm_only=True` on `EmrApplicationUsageRequest` and pass a persistent
+`cache_dir` to `calculate_emr_application_usage`. In this opt-in mode:
+
+- Only RM objects are downloaded. An explicit `yarn_log_uri` is authoritative:
+  missing uploads return retryable pending evidence, with no fallback to EMR/NM
+  archives. Without it, RM objects are selected from the cluster's `LogUri`.
+- The Spark event log is not read until an attempt has observed the target
+  application's RM summary. A source/application-scoped scheduling hint then
+  gives event downloads and parsing priority on subsequent attempts, so repeated
+  active-RM refreshes cannot starve their checkpoints. The hint is not evidence:
+  every complete result still refreshes and validates the current RM snapshot.
+  A missing RM archive still returns pending before reading event logs. Priority
+  attempts stop that availability probe at the first RM object, then list the
+  full RM snapshot after event processing, immediately before index refresh.
+  This avoids carrying pre-event-work ETags across slow event processing while
+  preserving conditional GETs and concurrent-refresh checks.
+  Only the target application's indexed records and historical
+  scheduler/node metadata are passed to the accounting parser.
+- A disposable SQLite index per bucket/prefix reuses objects with unchanged
+  ETag, size, and modification time. ETags are opaque identities. Downloads use
+  conditional GETs; replacements commit atomically per object. Deleted objects
+  are removed from the current snapshot. Objects without ETags are not reused.
+- Never-indexed RM objects are processed before refreshing previously indexed
+  objects, so a growing active log cannot repeatedly displace cold archive work.
+- With `cache_dir`, complete S3 Spark event-log objects are published atomically
+  to an identity-keyed cache. Parsing checkpoints retain successful whole-segment
+  metadata before final validation. A changed segment invalidates all subsequent
+  checkpoints; failed reads are never checkpointed. Local files also support
+  parsing checkpoints, identified by device, inode, size, and nanosecond mtime.
+  Do not mutate local input files while accounting is running.
+- Timeouts, concurrent index refresh conflicts, S3 replacement/deletion races,
+  and unreadable compressed RM archives return retryable pending evidence, never
+  complete results from a partial refresh. Completed objects survive retries.
+
+`timeout_seconds` defaults to 120 and must be finite and positive. This is a
+cooperative processing budget, **not a hard wall-clock deadline**: it cannot
+interrupt an in-flight SDK request or decompression operation. Configure bounded
+connect/read timeouts and retries on the injected clients. A single object must
+fit within the budget to be indexed; use smaller rolled logs or a larger budget
+if cold attempts repeatedly stop on the same object. Spark event-log downloads
+and parsing checkpoint separately: each download and each segment's parsing
+must individually fit the budget, but their aggregate need not fit one attempt.
+Resume requires the same persistent `cache_dir`; without it attempts are stateless.
+Once event metadata is warm, the current mutable RM snapshot plus listing and
+warm-cache lookup overhead must still fit one attempt. Multiple changing RM
+objects whose combined refresh exceeds the budget require smaller rolled logs
+or a larger budget; per-object caching cannot make that snapshot converge.
+
+The index requires a private local filesystem with SQLite locking; do not share
+it through object storage or a network filesystem. Multiple local processes may
+use it; lock contention returns pending after a short wait. It stores normalized
+log evidence, with owner-only permissions on newly created databases/directories.
+SQLite uses WAL with NORMAL synchronization: transactions remain atomic, but
+host power loss can discard recent cache commits and cause extra downloads.
+The cache also retains full Spark event-log objects and JSON metadata checkpoints,
+which can require substantially more space than the RM index alone.
+It is not an authoritative accounting store. No cross-source TTL or size-based
+eviction is performed: provision a quota/monitor disk usage, and remove obsolete
+source indexes only when their callers are stopped. Removing an index is safe
+then, but the next call incurs a cold scan. Deleted-object pages are reused by
+SQLite; the database file does not automatically shrink.
+
+Cold calls still read the RM history. Warm calls still list the source prefix
+and read global metadata; the optimization avoids repeated archive downloads
+and parsing unrelated applications, not all work proportional to cluster age.
+Use the narrow shipper prefix to reduce listing overhead. INFO events
+`yarn_archive_list`, `yarn_archive_download`, `yarn_rm_index`,
+`yarn_application_parse`, `yarn_eventlog_materialize`, and `yarn_eventlog_parse`
+report phase timing and, where applicable, objects, bytes, and cache reuse.
+
 `event_log_uri` accepts an S3 URI, a plain local path, or a local `file://` URI.
 Passing a pre-materialized local file or rolling-event-log directory avoids an
 S3 download; the selected event log is still streamed to extract accounting

yarn-resource-cost/package_yarn_job_cost.py

@@ -25,12 +25,14 @@
     PROJECT_DIR / "pyproject.toml": "pyproject.toml",
     PROJECT_DIR / "yarn_resource_cost.py": "yarn_resource_cost.py",
     PROJECT_DIR / "yarn_job_cost_api.py": "yarn_job_cost_api.py",
+    PROJECT_DIR / "yarn_job_cost_checkpoint.py": "yarn_job_cost_checkpoint.py",
     PROJECT_DIR / "yarn_job_cost_core.py": "yarn_job_cost_core.py",
     PROJECT_DIR / "yarn_job_cost_adapters.py": "yarn_job_cost_adapters.py",
     PROJECT_DIR / "yarn_job_cost_dataproc.py": "yarn_job_cost_dataproc.py",
     PROJECT_DIR / "calculate_yarn_job_cost.py": "calculate_yarn_job_cost.py",
     PROJECT_DIR / "yarn_job_cost_discovery.py": "yarn_job_cost_discovery.py",
     PROJECT_DIR / "yarn_job_cost_eventlog.py": "yarn_job_cost_eventlog.py",
+    PROJECT_DIR / "yarn_job_cost_index.py": "yarn_job_cost_index.py",
     PROJECT_DIR / "yarn_job_cost_defaults.py": "yarn_job_cost_defaults.py",
     PROJECT_DIR / "test_calculate_yarn_job_cost.py": "test_calculate_yarn_job_cost.py",
     PROJECT_DIR / "test_portable_yarn_resource_cost.py": "test_portable_yarn_resource_cost.py",

yarn-resource-cost/pyproject.toml

@@ -28,10 +28,12 @@ py-modules = [
   "calculate_yarn_job_cost",
   "yarn_job_cost_adapters",
   "yarn_job_cost_api",
+  "yarn_job_cost_checkpoint",
   "yarn_job_cost_core",
   "yarn_job_cost_dataproc",
   "yarn_job_cost_defaults",
   "yarn_job_cost_discovery",
   "yarn_job_cost_eventlog",
+  "yarn_job_cost_index",
   "yarn_resource_cost",
 ]