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 5 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
127 changes: 127 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,133 @@ 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_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 "
"<memory:8192, vCores:8, yarn.io/gpu:1>\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 "
"<memory:2048, vCores:2, yarn.io/gpu:1>\n"
"2026-01-01 00:00:02,000 INFO X: Registered with ResourceManager "
"as worker-b:8041 with total resource of "
"<memory:16384, vCores:16, yarn.io/gpu:2>\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 "
"<memory:8192, vCores:8, yarn.io/gpu:1>\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 "
"<memory:16384, vCores:16, yarn.io/gpu:2>\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 "
"<memory:2048, vCores:2, 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("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"
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
48 changes: 38 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_hosts_by_dir: dict[Path, set[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 = 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)
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_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)

@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,13 @@ 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 node_id == path_node_id:

@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 track whether this file has reported a host instead of comparing node_id with the directory name? If the directory is worker-a, a later worker-b registration can relabel A’s container while leaving A’s capacity on it.

@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. NodeManager parsing now tracks an explicit reported host per file rather than comparing the mutable node_id with the directory name, so a later registration for another worker cannot relabel an already-attributed container. Added a regression for that sequence. Leaving this open for your review.

unresolved_nm_containers.setdefault(file, []).append(container)
continue
done = DONE_RE.search(line)
if done:
Expand All @@ -361,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.
Expand Down
Loading