From 70aca844cd4a40b090e98811747053c22e4ee47e Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:59:10 +0000 Subject: [PATCH 1/8] Fix worker identity in partial YARN logs Signed-off-by: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> --- .../test_portable_yarn_resource_cost.py | 46 +++++++++++++++++++ yarn-resource-cost/yarn_job_cost_core.py | 22 +++++---- 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/yarn-resource-cost/test_portable_yarn_resource_cost.py b/yarn-resource-cost/test_portable_yarn_resource_cost.py index f8df7c3c..cdd0d672 100644 --- a/yarn-resource-cost/test_portable_yarn_resource_cost.py +++ b/yarn-resource-cost/test_portable_yarn_resource_cost.py @@ -72,6 +72,52 @@ def test_fair_scheduler_drf_policy_is_detected_from_evidence(self): self.assertEqual("DominantResourceCalculator", evidence.calculator_class) self.assertEqual("DominantResourceFairnessPolicy", evidence.scheduler_policy) + def test_nodemanager_registration_preserves_reported_worker_identity(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) / "hadoop-yarn" + log_dir.mkdir() + path = log_dir / "hadoop-yarn-nodemanager.log" + path.write_text( + "2026-01-01 00:00:00,000 INFO X: Registered with ResourceManager " + "as worker1:8041 with total resource of " + "\n" + "2026-01-01 00:00:01,000 INFO X: Start request for " + "container_1_0001_01_000002 with resource " + "\n", + encoding="utf-8", + ) + evidence = core.parse_yarn_logs(log_dir) + + node = evidence.nodes["worker1"] + container = evidence.containers["container_1_0001_01_000002"] + self.assertEqual(8192, node.memory_mb) + self.assertEqual(1, node.resources["yarn.io/gpu"]) + self.assertEqual("worker1", container.node_id) + self.assertEqual(8192, container.node_memory_mb) + + def test_resourcemanager_assignment_preserves_host_without_registration(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) / "hadoop-yarn" + log_dir.mkdir() + path = log_dir / "hadoop-yarn-resourcemanager.log" + path.write_text( + "2026-01-01 00:00:01,000 INFO X: Assigned container " + "container_1_0001_01_000002 of capacity " + " " + "on host worker1:8041\n", + encoding="utf-8", + ) + evidence = core.parse_yarn_logs(log_dir) + + node = evidence.nodes["worker1"] + container = evidence.containers["container_1_0001_01_000002"] + self.assertIsNone(node.memory_mb) + self.assertIsNone(node.vcores) + self.assertEqual({}, node.resources) + self.assertEqual("worker1", container.node_id) + self.assertEqual(0, container.node_memory_mb) + self.assertEqual(0, container.node_vcores) + class AdapterTest(unittest.TestCase): def test_catalog_cost_and_node_mapping(self): diff --git a/yarn-resource-cost/yarn_job_cost_core.py b/yarn-resource-cost/yarn_job_cost_core.py index e89308f9..cbfee7ca 100644 --- a/yarn-resource-cost/yarn_job_cost_core.py +++ b/yarn-resource-cost/yarn_job_cost_core.py @@ -42,7 +42,7 @@ r"totalAllocatedContainers=(?P\d+)" ) NODE_RE = re.compile( - r"Registered with ResourceManager .* total resource of " + r"Registered with ResourceManager(?: as (?P[^:,\s]+):\d+)? .* total resource of " r"\d+), vCores:(?P\d+)(?P[^>]*)>" ) RM_NODE_RE = re.compile( @@ -263,7 +263,8 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: for file in files: path_node_id = file.parent.name - path_node = evidence.nodes.setdefault(path_node_id, Node(path_node_id)) + node_id = path_node_id + node = evidence.nodes.setdefault(node_id, Node(node_id)) with open_log(file) as handle: for raw_line in handle: line = normalize_log_line(raw_line) @@ -324,17 +325,20 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: int(node_match.group("vcores")), node_match.group("resources"), ) - path_node.instance_type = _node_instance_type(line) - path_node.memory_mb = resources["memory-mb"] - path_node.vcores = resources["vcores"] - path_node.gpus = resources.get("yarn.io/gpu", 0) - path_node.resources = resources + node_id = node_match.group("host") or path_node_id + node = evidence.nodes.setdefault(node_id, Node(node_id)) + node.instance_type = _node_instance_type(line) + node.memory_mb = resources["memory-mb"] + node.vcores = resources["vcores"] + node.gpus = resources.get("yarn.io/gpu", 0) + node.resources = resources continue assignment = RM_ASSIGN_RE.search(line) if assignment: host = assignment.group("host") + assigned_node = evidence.nodes.setdefault(host, Node(host)) candidate = _container_from_match( - assignment, host, evidence.nodes.get(host), "resourcemanager" + assignment, host, assigned_node, "resourcemanager" ) evidence.containers[candidate.container_id] = candidate continue @@ -349,7 +353,7 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: start = START_RE.search(line) if start: candidate = _container_from_match( - start, path_node_id, path_node, "nodemanager" + start, node_id, node, "nodemanager" ) evidence.containers.setdefault(candidate.container_id, candidate) continue From 44e9bb70a81f27bb93a123703305fbc98beefa16 Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Fri, 2 Oct 2026 01:18:37 +0500 Subject: [PATCH 2/8] Fix NodeManager identity across split log files --- yarn-resource-cost/yarn_job_cost_core.py | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/yarn-resource-cost/yarn_job_cost_core.py b/yarn-resource-cost/yarn_job_cost_core.py index cbfee7ca..3078755e 100644 --- a/yarn-resource-cost/yarn_job_cost_core.py +++ b/yarn-resource-cost/yarn_job_cost_core.py @@ -257,13 +257,15 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: evidence = YarnEvidence() rm_finishes: dict[str, int] = {} nm_finishes: dict[str, int] = {} + nm_node_ids: dict[Path, str] = {} + unresolved_nm_containers: dict[Path, list[Container]] = {} files = relevant_log_files(path) if not files: raise ValueError(f"No ResourceManager or NodeManager log files found under {path}") for file in files: path_node_id = file.parent.name - node_id = path_node_id + node_id = nm_node_ids.get(file.parent, path_node_id) node = evidence.nodes.setdefault(node_id, Node(node_id)) with open_log(file) as handle: for raw_line in handle: @@ -327,6 +329,10 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: ) node_id = node_match.group("host") or path_node_id node = evidence.nodes.setdefault(node_id, Node(node_id)) + if node_match.group("host"): + nm_node_ids[file.parent] = node_id + for container in unresolved_nm_containers.pop(file.parent, []): + container.node_id = node_id node.instance_type = _node_instance_type(line) node.memory_mb = resources["memory-mb"] node.vcores = resources["vcores"] @@ -355,7 +361,15 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: candidate = _container_from_match( start, node_id, node, "nodemanager" ) - evidence.containers.setdefault(candidate.container_id, candidate) + container = evidence.containers.setdefault( + candidate.container_id, candidate + ) + if ( + container is candidate + and file.parent not in nm_node_ids + and node_id == path_node_id + ): + unresolved_nm_containers.setdefault(file.parent, []).append(container) continue done = DONE_RE.search(line) if done: From 2ea6867c94d65558f5c36f9daf9cadb588cdd054 Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Fri, 2 Oct 2026 01:18:39 +0500 Subject: [PATCH 3/8] Test split NodeManager log identity --- .../test_portable_yarn_resource_cost.py | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/yarn-resource-cost/test_portable_yarn_resource_cost.py b/yarn-resource-cost/test_portable_yarn_resource_cost.py index cdd0d672..8a760c97 100644 --- a/yarn-resource-cost/test_portable_yarn_resource_cost.py +++ b/yarn-resource-cost/test_portable_yarn_resource_cost.py @@ -95,6 +95,30 @@ def test_nodemanager_registration_preserves_reported_worker_identity(self): self.assertEqual("worker1", container.node_id) self.assertEqual(8192, container.node_memory_mb) + def test_nodemanager_identity_is_preserved_across_split_log_files(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) / "hadoop-yarn" + log_dir.mkdir() + (log_dir / "00-start.log").write_text( + "2026-01-01 00:00:01,000 INFO X: Start request for " + "container_1_0001_01_000002 with resource " + "\n", + encoding="utf-8", + ) + (log_dir / "01-registration.log").write_text( + "2026-01-01 00:00:00,000 INFO X: Registered with ResourceManager " + "as worker1:8041 with total resource of " + "\n", + encoding="utf-8", + ) + evidence = core.parse_yarn_logs(log_dir) + + container = evidence.containers["container_1_0001_01_000002"] + self.assertEqual("worker1", container.node_id) + self.assertEqual(8192, container.node_memory_mb) + self.assertEqual(8, container.node_vcores) + self.assertEqual(1, container.node_gpus) + def test_resourcemanager_assignment_preserves_host_without_registration(self): with tempfile.TemporaryDirectory() as directory: log_dir = Path(directory) / "hadoop-yarn" From e589f5fd9f6c0cea9d6d124535504243c4cee50f Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Fri, 2 Oct 2026 01:29:07 +0500 Subject: [PATCH 4/8] Fix worker identity inference for co-located logs --- yarn-resource-cost/yarn_job_cost_core.py | 30 ++++++++++++++++-------- 1 file changed, 20 insertions(+), 10 deletions(-) diff --git a/yarn-resource-cost/yarn_job_cost_core.py b/yarn-resource-cost/yarn_job_cost_core.py index 3078755e..937c3f85 100644 --- a/yarn-resource-cost/yarn_job_cost_core.py +++ b/yarn-resource-cost/yarn_job_cost_core.py @@ -257,7 +257,7 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: evidence = YarnEvidence() rm_finishes: dict[str, int] = {} nm_finishes: dict[str, int] = {} - nm_node_ids: dict[Path, str] = {} + nm_hosts_by_dir: dict[Path, set[str]] = {} unresolved_nm_containers: dict[Path, list[Container]] = {} files = relevant_log_files(path) if not files: @@ -265,7 +265,7 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: for file in files: path_node_id = file.parent.name - node_id = nm_node_ids.get(file.parent, path_node_id) + node_id = path_node_id node = evidence.nodes.setdefault(node_id, Node(node_id)) with open_log(file) as handle: for raw_line in handle: @@ -330,8 +330,8 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: node_id = node_match.group("host") or path_node_id node = evidence.nodes.setdefault(node_id, Node(node_id)) if node_match.group("host"): - nm_node_ids[file.parent] = node_id - for container in unresolved_nm_containers.pop(file.parent, []): + nm_hosts_by_dir.setdefault(file.parent, set()).add(node_id) + for container in unresolved_nm_containers.pop(file, []): container.node_id = node_id node.instance_type = _node_instance_type(line) node.memory_mb = resources["memory-mb"] @@ -364,12 +364,8 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: container = evidence.containers.setdefault( candidate.container_id, candidate ) - if ( - container is candidate - and file.parent not in nm_node_ids - and node_id == path_node_id - ): - unresolved_nm_containers.setdefault(file.parent, []).append(container) + if container is candidate and node_id == path_node_id: + unresolved_nm_containers.setdefault(file, []).append(container) continue done = DONE_RE.search(line) if done: @@ -379,6 +375,20 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: finish, nm_finishes.get(container_id, finish) ) + # A rotated single-worker archive can put registration and starts in + # different files. Infer across files only when the directory contains one + # unambiguous reported NodeManager host. With multiple workers, keep the + # path-derived identity unknown rather than assigning another worker's + # capacity. + for unresolved_file, containers in unresolved_nm_containers.items(): + hosts = nm_hosts_by_dir.get(unresolved_file.parent, set()) + if len(hosts) != 1: + continue + host = next(iter(hosts)) + for container in containers: + if container.node_id == unresolved_file.parent.name: + container.node_id = host + for container_id, container in evidence.containers.items(): # Registration evidence may be stored in a later file than the # allocation. Reconcile containers after every archived log was parsed. From f87e3ee17125f2391cd923abb580a2db6a25b818 Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Fri, 2 Oct 2026 01:29:22 +0500 Subject: [PATCH 5/8] Test co-located NodeManager identity isolation --- .../test_portable_yarn_resource_cost.py | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/yarn-resource-cost/test_portable_yarn_resource_cost.py b/yarn-resource-cost/test_portable_yarn_resource_cost.py index 8a760c97..0ab4bd4a 100644 --- a/yarn-resource-cost/test_portable_yarn_resource_cost.py +++ b/yarn-resource-cost/test_portable_yarn_resource_cost.py @@ -119,6 +119,63 @@ def test_nodemanager_identity_is_preserved_across_split_log_files(self): self.assertEqual(8, container.node_vcores) self.assertEqual(1, container.node_gpus) + def test_nodemanager_identity_does_not_cross_workers_in_same_directory(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) / "hadoop-yarn" + log_dir.mkdir() + (log_dir / "00-worker-a.log").write_text( + "2026-01-01 00:00:00,000 INFO X: Registered with ResourceManager " + "as worker-a:8041 with total resource of " + "\n", + encoding="utf-8", + ) + (log_dir / "01-worker-b.log").write_text( + "2026-01-01 00:00:01,000 INFO X: Start request for " + "container_1_0001_01_000002 with resource " + "\n" + "2026-01-01 00:00:02,000 INFO X: Registered with ResourceManager " + "as worker-b:8041 with total resource of " + "\n", + encoding="utf-8", + ) + evidence = core.parse_yarn_logs(log_dir) + + container = evidence.containers["container_1_0001_01_000002"] + self.assertEqual("worker-b", container.node_id) + self.assertEqual(16384, container.node_memory_mb) + self.assertEqual(16, container.node_vcores) + self.assertEqual(2, container.node_gpus) + + def test_ambiguous_split_nodemanager_logs_do_not_guess_worker_identity(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) / "hadoop-yarn" + log_dir.mkdir() + (log_dir / "00-worker-a.log").write_text( + "2026-01-01 00:00:00,000 INFO X: Registered with ResourceManager " + "as worker-a:8041 with total resource of " + "\n", + encoding="utf-8", + ) + (log_dir / "01-worker-b.log").write_text( + "2026-01-01 00:00:00,000 INFO X: Registered with ResourceManager " + "as worker-b:8041 with total resource of " + "\n", + encoding="utf-8", + ) + (log_dir / "02-start.log").write_text( + "2026-01-01 00:00:01,000 INFO X: Start request for " + "container_1_0001_01_000002 with resource " + "\n", + encoding="utf-8", + ) + evidence = core.parse_yarn_logs(log_dir) + + container = evidence.containers["container_1_0001_01_000002"] + self.assertEqual("hadoop-yarn", container.node_id) + self.assertEqual(0, container.node_memory_mb) + self.assertEqual(0, container.node_vcores) + self.assertEqual(0, container.node_gpus) + def test_resourcemanager_assignment_preserves_host_without_registration(self): with tempfile.TemporaryDirectory() as directory: log_dir = Path(directory) / "hadoop-yarn" From 16c67111fe31f55bd1dd787adc91ffcc8a30fa11 Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Fri, 2 Oct 2026 17:14:19 +0500 Subject: [PATCH 6/8] fix(yarn): preserve registration evidence and per-file host identity --- yarn-resource-cost/yarn_job_cost_core.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/yarn-resource-cost/yarn_job_cost_core.py b/yarn-resource-cost/yarn_job_cost_core.py index 937c3f85..a65785e0 100644 --- a/yarn-resource-cost/yarn_job_cost_core.py +++ b/yarn-resource-cost/yarn_job_cost_core.py @@ -267,6 +267,7 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: path_node_id = file.parent.name node_id = path_node_id node = evidence.nodes.setdefault(node_id, Node(node_id)) + file_reported_host: str | None = None with open_log(file) as handle: for raw_line in handle: line = normalize_log_line(raw_line) @@ -327,17 +328,22 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: int(node_match.group("vcores")), node_match.group("resources"), ) - node_id = node_match.group("host") or path_node_id + reported_host = node_match.group("host") + node_id = reported_host or path_node_id node = evidence.nodes.setdefault(node_id, Node(node_id)) - if node_match.group("host"): + if reported_host: + file_reported_host = node_id nm_hosts_by_dir.setdefault(file.parent, set()).add(node_id) for container in unresolved_nm_containers.pop(file, []): container.node_id = node_id - node.instance_type = _node_instance_type(line) + instance_type = _node_instance_type(line) + if instance_type: + node.instance_type = instance_type node.memory_mb = resources["memory-mb"] node.vcores = resources["vcores"] node.gpus = resources.get("yarn.io/gpu", 0) node.resources = resources + node.registered_with_rm = True continue assignment = RM_ASSIGN_RE.search(line) if assignment: @@ -364,7 +370,7 @@ def parse_yarn_logs(path: Path) -> YarnEvidence: container = evidence.containers.setdefault( candidate.container_id, candidate ) - if container is candidate and node_id == path_node_id: + if container is candidate and file_reported_host is None: unresolved_nm_containers.setdefault(file, []).append(container) continue done = DONE_RE.search(line) From 9d9f79bf47317a979c80d849b1dfad895644bd1d Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Fri, 2 Oct 2026 17:14:31 +0500 Subject: [PATCH 7/8] fix(yarn): require node registration evidence for complete accounting --- yarn-resource-cost/calculate_yarn_job_cost.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/yarn-resource-cost/calculate_yarn_job_cost.py b/yarn-resource-cost/calculate_yarn_job_cost.py index 2532e4f0..1fd15005 100644 --- a/yarn-resource-cost/calculate_yarn_job_cost.py +++ b/yarn-resource-cost/calculate_yarn_job_cost.py @@ -733,6 +733,22 @@ def calculate_applications( "NodeManager DONE fallback" ) + unregistered_node_ids = sorted( + { + container.node_id + for container in complete_containers + if ( + evidence.nodes.get(container.node_id) is None + or not evidence.nodes[container.node_id].registered_with_rm + ) + } + ) + if unregistered_node_ids: + warnings.append( + "Node registration evidence is missing for: " + + ", ".join(unregistered_node_ids) + ) + by_instance_type: dict[str, float] = defaultdict(float) by_instance_vcore_seconds: dict[str, float] = defaultdict(float) container_seconds = 0.0 @@ -786,6 +802,7 @@ def calculate_applications( or nm_finish_fallbacks > 0 or unknown_instance_type or missing_gpu_capacity + or bool(unregistered_node_ids) or bool(resource_capacity_errors) ) permanent_incomplete_evidence = evidence.accounting_policy_ambiguous From b0920a5026db87a98ce5e5c82868c124c8357b97 Mon Sep 17 00:00:00 2001 From: mnkj0021 <98058999+mnkj0021@users.noreply.github.com> Date: Fri, 2 Oct 2026 17:14:52 +0500 Subject: [PATCH 8/8] test(yarn): cover registration and host-identity review cases --- .../test_portable_yarn_resource_cost.py | 82 +++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/yarn-resource-cost/test_portable_yarn_resource_cost.py b/yarn-resource-cost/test_portable_yarn_resource_cost.py index 0ab4bd4a..4b4fde0e 100644 --- a/yarn-resource-cost/test_portable_yarn_resource_cost.py +++ b/yarn-resource-cost/test_portable_yarn_resource_cost.py @@ -11,6 +11,7 @@ import unittest from pathlib import Path +import calculate_yarn_job_cost as reporting import yarn_job_cost_adapters as adapters import yarn_job_cost_core as core @@ -92,6 +93,7 @@ def test_nodemanager_registration_preserves_reported_worker_identity(self): container = evidence.containers["container_1_0001_01_000002"] self.assertEqual(8192, node.memory_mb) self.assertEqual(1, node.resources["yarn.io/gpu"]) + self.assertTrue(node.registered_with_rm) self.assertEqual("worker1", container.node_id) self.assertEqual(8192, container.node_memory_mb) @@ -176,6 +178,85 @@ def test_ambiguous_split_nodemanager_logs_do_not_guess_worker_identity(self): self.assertEqual(0, container.node_vcores) self.assertEqual(0, container.node_gpus) + def test_explicit_file_host_is_not_relabelled_by_later_registration(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) / "worker-a" + log_dir.mkdir() + path = log_dir / "hadoop-yarn-nodemanager.log" + path.write_text( + "2026-01-01 00:00:00,000 INFO X: Registered with ResourceManager " + "as worker-a:8041 with total resource of " + "\n" + "2026-01-01 00:00:01,000 INFO X: Start request for " + "container_1_0001_01_000002 with resource " + "\n" + "2026-01-01 00:00:02,000 INFO X: Registered with ResourceManager " + "as worker-b:8041 with total resource of " + "\n", + encoding="utf-8", + ) + evidence = core.parse_yarn_logs(log_dir) + + container = evidence.containers["container_1_0001_01_000002"] + self.assertEqual("worker-a", container.node_id) + self.assertEqual(8192, container.node_memory_mb) + self.assertEqual(8, container.node_vcores) + + def test_nodemanager_registration_preserves_existing_instance_type(self): + with tempfile.TemporaryDirectory() as directory: + log_dir = Path(directory) / "hadoop-yarn" + log_dir.mkdir() + path = log_dir / "hadoop-yarn.log" + path.write_text( + "2026-01-01 00:00:00,000 INFO RMNodeImpl: NodeManager from node " + "worker1(cmPort: 8041 httpPort: 8042) registered with capability: " + " instanceType(STRING)=g4dn.4xlarge\n" + "2026-01-01 00:00:01,000 INFO X: Registered with ResourceManager " + "as worker1:8041 with total resource of \n", + encoding="utf-8", + ) + evidence = core.parse_yarn_logs(log_dir) + + self.assertEqual("g4dn.4xlarge", evidence.nodes["worker1"].instance_type) + + def test_node_class_map_does_not_make_unregistered_host_complete(self): + container = self.container( + source="resourcemanager", + finish_source="resourcemanager", + ) + node = core.Node(node_id="worker1") + evidence = core.YarnEvidence( + nodes={"worker1": node}, + containers={container.container_id: container}, + application_summaries={ + "application_1_0001": core.ApplicationSummary( + application_id="application_1_0001", + name="test", + final_status="SUCCEEDED", + total_allocated_containers=1, + ) + }, + ) + with tempfile.TemporaryDirectory() as directory: + mapping = Path(directory) / "node-classes.json" + mapping.write_text( + json.dumps({"nodes": {"worker1": "onprem:worker-8"}}), + encoding="utf-8", + ) + adapters.apply_node_class_map(evidence, mapping) + + application = reporting.calculate_applications( + evidence, + "default", + {}, + include_application_master=False, + )[0] + self.assertEqual("onprem:worker-8", node.node_class) + self.assertFalse(application["complete"]) + self.assertTrue( + any("Node registration evidence is missing" in w for w in application["warnings"]) + ) + def test_resourcemanager_assignment_preserves_host_without_registration(self): with tempfile.TemporaryDirectory() as directory: log_dir = Path(directory) / "hadoop-yarn" @@ -195,6 +276,7 @@ def test_resourcemanager_assignment_preserves_host_without_registration(self): self.assertIsNone(node.memory_mb) self.assertIsNone(node.vcores) self.assertEqual({}, node.resources) + self.assertFalse(node.registered_with_rm) self.assertEqual("worker1", container.node_id) self.assertEqual(0, container.node_memory_mb) self.assertEqual(0, container.node_vcores)