Add application-scoped YARN resource usage API

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

@@ -29,5 +29,29 @@ jobs:
         run: python -m py_compile yarn-resource-cost/*.py
       - name: Unit tests
         run: python -m unittest discover -s yarn-resource-cost -p 'test*.py'
+      - name: Installed console error handling
+        run: |
+          python -m pip install ./yarn-resource-cost
+          python - <<'PY'
+          import shutil
+          import subprocess
+
+          command = shutil.which("yarn-resource-cost")
+          assert command is not None
+          completed = subprocess.run(
+              [
+                  command,
+                  "--adapter",
+                  "on-prem",
+                  "--event-log-root",
+                  "/definitely/not/a/yarn/event/log",
+              ],
+              capture_output=True,
+              text=True,
+          )
+          assert completed.returncode == 2, completed
+          assert completed.stderr.startswith("error: "), completed.stderr
+          assert "Traceback" not in completed.stderr, completed.stderr
+          PY
       - name: Standalone bundle smoke test
         run: python yarn-resource-cost/package_yarn_job_cost.py --output-dir /tmp/yarn-resource-cost-dist

yarn-resource-cost/README.md

@@ -38,6 +38,50 @@ DominantResourceCalculator:
 NodeEquivalentSeconds = ContainerSeconds * NodeShare
 ```
 
+## Python API
+
+Install the subproject and its optional AWS dependency:
+
+```bash
+python3 -m pip install './yarn-resource-cost[aws]'
+```
+
+The application-scoped API accepts injected boto3 clients and returns resource
+usage without imposing a pricing policy:
+
+```python
+import boto3
+
+from yarn_resource_cost import (
+    EmrApplicationUsageRequest,
+    calculate_emr_application_usage,
+)
+
+session = boto3.Session(region_name="us-west-2")
+usage = calculate_emr_application_usage(
+    EmrApplicationUsageRequest(
+        cluster_id="j-EXAMPLE",
+        application_id="application_123_0001",
+        event_log_uri="s3://example-bucket/spark-events/eventlog_v2_application_123_0001/",
+        region="us-west-2",
+    ),
+    emr_client=session.client("emr"),
+    s3_client=session.client("s3"),
+)
+print(usage.instance_seconds_by_type)
+```
+
+`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
+metadata.
+
+Incomplete archived logs return `complete=False` and indicate whether a later
+retry can help. Missing summaries, allocations, terminal transitions, or node
+registration metadata are retryable; contradictory or unsupported accounting
+policies are not. Authentication and transport errors propagate from boto3.
+The caller decides whether and how to translate instance-seconds into currency.
+
 Memory, vcores, `yarn.io/gpu`, and arbitrary numeric custom resources are
 parsed generically. Heterogeneous node classes remain separate in structured
 output and expressions such as:

yarn-resource-cost/calculate_yarn_job_cost.py

@@ -754,6 +754,17 @@ def calculate_applications(
         )
         if unknown_instance_type:
             warnings.append("One or more allocated containers have an unknown instance type")
+        transient_incomplete_evidence = (
+            incomplete > 0
+            or not coverage_complete
+            or nm_start_fallbacks > 0
+            or nm_finish_fallbacks > 0
+            or unknown_instance_type
+            or missing_gpu_capacity
+            or bool(resource_capacity_errors)
+        )
+        permanent_incomplete_evidence = evidence.accounting_policy_ambiguous
+        complete = not transient_incomplete_evidence and not permanent_incomplete_evidence
         starts = [container.start_ms for container in containers]
         finishes = [
             container.finish_ms for container in complete_containers if container.finish_ms is not None
@@ -789,16 +800,11 @@ def calculate_applications(
             "cost_expression": expression,
             "first_container_start_utc": iso_utc(min(starts) if starts else None),
             "last_container_finish_utc": iso_utc(max(finishes) if finishes else None),
-            "complete": (
-                incomplete == 0
-                and coverage_complete
-                and nm_start_fallbacks == 0
-                and nm_finish_fallbacks == 0
-                and not unknown_instance_type
-                and not missing_gpu_capacity
-                and not resource_capacity_errors
-                and not evidence.accounting_policy_ambiguous
-            ),
+            "complete": complete,
+            # A fresh archive snapshot can resolve missing summaries, allocations,
+            # terminal transitions, and incomplete node registration metadata. It
+            # cannot resolve conflicting accounting-policy evidence already present.
+            "retryable": not complete and not permanent_incomplete_evidence,
             "warnings": warnings,
         }
         results.append(result)

yarn-resource-cost/package_yarn_job_cost.py

@@ -22,7 +22,9 @@
 PROJECT_DIR = Path(__file__).resolve().parent
 DEFAULT_LICENSE_FILE = PROJECT_DIR.parent / "LICENSE"
 PACKAGE_SOURCES = {
+    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_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",
@@ -36,6 +38,7 @@
     PROJECT_DIR / "test_dataproc_log_normalization.py": "test_dataproc_log_normalization.py",
     PROJECT_DIR / "test_dataproc_adapter.py": "test_dataproc_adapter.py",
     PROJECT_DIR / "test_portable_comparison.py": "test_portable_comparison.py",
+    PROJECT_DIR / "test_yarn_job_cost_api.py": "test_yarn_job_cost_api.py",
     PROJECT_DIR / "RESOURCE_COST_MODEL.md": "RESOURCE_COST_MODEL.md",
     PROJECT_DIR / "tests/fixtures/on_prem/eventlog_v2_application_1_0001/events_1_application_1_0001": "tests/fixtures/on_prem/eventlog_v2_application_1_0001/events_1_application_1_0001",
     PROJECT_DIR / "tests/fixtures/on_prem/yarn/hadoop-yarn-resourcemanager-rm.log": "tests/fixtures/on_prem/yarn/hadoop-yarn-resourcemanager-rm.log",

yarn-resource-cost/pyproject.toml

@@ -0,0 +1,36 @@
+[build-system]
+requires = ["setuptools>=77", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "yarn-resource-cost"
+version = "0.1.0"
+description = "Portable YARN resource accounting for Spark applications"
+readme = "README.md"
+requires-python = ">=3.10,<3.13"
+license = "Apache-2.0"
+classifiers = [
+  "Programming Language :: Python :: 3",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+]
+
+[project.optional-dependencies]
+aws = ["boto3>=1.34"]
+
+[project.scripts]
+yarn-resource-cost = "yarn_resource_cost:cli_main"
+
+[tool.setuptools]
+py-modules = [
+  "calculate_yarn_job_cost",
+  "yarn_job_cost_adapters",
+  "yarn_job_cost_api",
+  "yarn_job_cost_core",
+  "yarn_job_cost_dataproc",
+  "yarn_job_cost_defaults",
+  "yarn_job_cost_discovery",
+  "yarn_job_cost_eventlog",
+  "yarn_resource_cost",
+]

yarn-resource-cost/test_calculate_yarn_job_cost.py

@@ -328,8 +328,38 @@ def test_rm_terminal_precedes_nm_done_and_nm_fallback_is_not_final(self):
             self.assertEqual(10.0, result["container_seconds"])
             self.assertEqual(1, result["nodemanager_finish_fallback_container_count"])
             self.assertFalse(result["complete"])
+            self.assertTrue(result["retryable"])
             self.assertIn("NodeManager DONE fallback", " | ".join(result["warnings"]))
 
+    def test_ambiguous_accounting_policy_is_not_retryable(self):
+        container = MODULE.Container(
+            container_id="container_123_0002_01_000002",
+            application_id=APP_ID,
+            node_id="worker",
+            start_ms=1000,
+            finish_ms=2000,
+            memory_mb=40,
+            node_memory_mb=100,
+            vcores=1,
+            node_vcores=4,
+            source="resourcemanager",
+            finish_source="resourcemanager",
+        )
+        summary = MODULE.ApplicationSummary(APP_ID, "test", "SUCCEEDED", 1)
+
+        evidence = MODULE.YarnEvidence(
+            nodes={"worker": MODULE.Node("worker", "cpu.test", 100, 4, 0)},
+            containers={container.container_id: container},
+            calculator_class="DefaultResourceCalculator",
+            accounting_policy_ambiguous=True,
+            application_summaries={APP_ID: summary},
+        )
+
+        result = MODULE.calculate_applications(evidence, "default", {}, False)[0]
+
+        self.assertFalse(result["complete"])
+        self.assertFalse(result["retryable"])
+
     def test_emr_log_cache_can_be_refreshed(self):
         with tempfile.TemporaryDirectory() as directory:
             cache = Path(directory)

yarn-resource-cost/test_portable_yarn_resource_cost.py

@@ -106,6 +106,26 @@ def test_missing_catalog_rate_suppresses_final_cost(self):
 
 
 class PortableCliTest(unittest.TestCase):
+    def test_entry_point_formats_expected_errors(self):
+        completed = subprocess.run(
+            [
+                sys.executable,
+                "-c",
+                "import yarn_resource_cost; yarn_resource_cost.cli_main()",
+                "--adapter",
+                "on-prem",
+                "--event-log-root",
+                "/definitely/not/a/yarn/event/log",
+            ],
+            cwd=ROOT,
+            capture_output=True,
+            text=True,
+        )
+
+        self.assertEqual(2, completed.returncode)
+        self.assertTrue(completed.stderr.startswith("error: "), completed.stderr)
+        self.assertNotIn("Traceback", completed.stderr)
+
     def test_on_prem_fixture_end_to_end_with_catalog(self):
         with tempfile.TemporaryDirectory() as directory:
             output = Path(directory) / "result.json"