Skip to content

Fix worker identity in partial YARN logs #279

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mnkj0021 wants to merge 8 commits into
base: dev
Choose a base branch
from
Open

Fix worker identity in partial YARN logs #279

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
70aca84
Fix worker identity in partial YARN logs
mnkj0021 Oct 1, 2026
44e9bb7
Fix NodeManager identity across split log files
mnkj0021 Oct 1, 2026
2ea6867
Test split NodeManager log identity
mnkj0021 Oct 1, 2026
e589f5f
Fix worker identity inference for co-located logs
mnkj0021 Oct 1, 2026
f87e3ee
Test co-located NodeManager identity isolation
mnkj0021 Oct 1, 2026
16c6711
fix(yarn): preserve registration evidence and per-file host identity
mnkj0021 Oct 2, 2026
9d9f79b
fix(yarn): require node registration evidence for complete accounting
mnkj0021 Oct 2, 2026
b0920a5
test(yarn): cover registration and host-identity review cases
mnkj0021 Oct 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 70 additions & 0 deletions yarn-resource-cost/test_portable_yarn_resource_cost.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,76 @@ 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 "
"<memory:8192, vCores:8, yarn.io/gpu:1>\n"
"2026-01-01 00:00:01,000 INFO X: Start request for "
"container_1_0001_01_000002 with resource "
"<memory:2048, vCores:2, yarn.io/gpu:1>\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_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 "
"<memory:2048, vCores:2, yarn.io/gpu:1>\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 "
"<memory:8192, vCores:8, yarn.io/gpu:1>\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"
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 "
"<memory:2048, vCores:2, yarn.io/gpu:1> "
"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):
Expand Down
38 changes: 28 additions & 10 deletions yarn-resource-cost/yarn_job_cost_core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@
r"totalAllocatedContainers=(?P<containers>\d+)"
)
NODE_RE = re.compile(
r"Registered with ResourceManager .* total resource of "
r"Registered with ResourceManager(?: as (?P<host>[^:,\s]+):\d+)? .* total resource of "
r"<memory:(?P<memory>\d+), vCores:(?P<vcores>\d+)(?P<resources>[^>]*)>"
)
RM_NODE_RE = re.compile(
Expand Down Expand Up @@ -257,13 +257,16 @@ 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
path_node = evidence.nodes.setdefault(path_node_id, Node(path_node_id))
node_id = nm_node_ids.get(file.parent, path_node_id)
Comment thread
mnkj0021 marked this conversation as resolved.
Resolved
Comment thread
mnkj0021 marked this conversation as resolved.
Resolved
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)
Expand Down Expand Up @@ -324,17 +327,24 @@ 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))
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved
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)

@gerashegalov gerashegalov Oct 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we preserve the existing instance type when this NM line doesn’t include one? Otherwise it clears a class already reported by the RM and can prevent pricing.

@mnkj0021 mnkj0021 Oct 2, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed on b0920a5. An NM registration now replaces instance_type only when that registration line actually reports one, so an RM-provided instance type is preserved otherwise. Added a focused regression. Leaving the thread open for confirmation.

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))

@gerashegalov gerashegalov Oct 2, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we keep the host in the node inventory without treating this placeholder as registration evidence? With assignment maxima and a node-class map, this now produces a final cost even though no node registration was logged.

@mnkj0021 mnkj0021 Oct 2, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed on the current head b0920a5. RM-assignment placeholders remain in the node inventory, but final accounting now treats a used node without registration evidence as incomplete even if a node-class map makes it priceable. I added regression coverage for the mapped-but-unregistered case. Leaving this thread open for your confirmation.

candidate = _container_from_match(
assignment, host, evidence.nodes.get(host), "resourcemanager"
assignment, host, assigned_node, "resourcemanager"
)
evidence.containers[candidate.container_id] = candidate
continue
Expand All @@ -349,9 +359,17 @@ 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)
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:
Expand Down
Loading