Skip to content
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

Less verbose log messages for remove and register worker #8938

Merged
merged 3 commits into from
Nov 15, 2024
Merged

Less verbose log messages for remove and register worker #8938

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
cc763c0
Less verbose log messages for remove and register worker
fjetter Nov 15, 2024
6504409
fix test
fjetter Nov 15, 2024
024f4eb
fix
fjetter Nov 15, 2024
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
14 changes: 9 additions & 5 deletions distributed/scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4571,7 +4571,7 @@ async def add_worker(
)
self.stimulus_queue_slots_maybe_opened(stimulus_id=stimulus_id)

logger.info("Register worker %s", ws)
logger.info("Register worker addr: %s name: %s", ws.address, ws.name)

msg = {
"status": "OK",
Expand Down Expand Up @@ -5428,7 +5428,9 @@ async def remove_worker(

ws = self.workers[address]

logger.info(f"Remove worker {ws} ({stimulus_id=})")
logger.info(
f"Remove worker addr: {ws.address} name: {ws.name} ({stimulus_id=})"
)
if close:
with suppress(AttributeError, CommClosedError):
self.stream_comms[address].send(
Expand Down Expand Up @@ -7548,7 +7550,11 @@ async def retire_workers(
names_set = {str(name) for name in names}
wss = {ws for ws in self.workers.values() if str(ws.name) in names_set}
elif workers is not None:
logger.info("Retire worker addresses %s", workers)
logger.info(
"Retire worker addresses (stimulus_id='%s') %s",
stimulus_id,
workers,
)
wss = {
self.workers[address]
for address in workers
Expand All @@ -7572,8 +7578,6 @@ async def retire_workers(
try:
coros = []
for ws in wss:
logger.info(f"Retiring worker {ws.address!r} ({stimulus_id=!r})")

policy = RetireWorker(ws.address)
amm.add_policy(policy)

Expand Down
12 changes: 4 additions & 8 deletions distributed/tests/test_worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2985,19 +2985,15 @@ async def test_log_remove_worker(c, s, a, b):

assert log.getvalue().splitlines() == [
# Successful graceful
f"Retire worker addresses ['{a.address}']",
f"Retiring worker '{a.address}' (stimulus_id='graceful')",
f"Remove worker <WorkerState '{a.address}', name: 0, status: "
"closing_gracefully, memory: 2, processing: 1> (stimulus_id='graceful')",
f"Retire worker addresses (stimulus_id='graceful') ['{a.address}']",
f"Remove worker addr: {a.address} name: {a.name} (stimulus_id='graceful')",
f"Retired worker '{a.address}' (stimulus_id='graceful')",
# Aborted graceful
f"Retire worker addresses ['{b.address}']",
f"Retiring worker '{b.address}' (stimulus_id='graceful_abort')",
f"Retire worker addresses (stimulus_id='graceful_abort') ['{b.address}']",
f"Could not retire worker '{b.address}': unique data could not be "
"moved to any other worker (stimulus_id='graceful_abort')",
# Ungraceful
f"Remove worker <WorkerState '{b.address}', name: 1, status: "
"running, memory: 2, processing: 1> (stimulus_id='ungraceful')",
f"Remove worker addr: {b.address} name: {b.name} (stimulus_id='ungraceful')",
f"Removing worker '{b.address}' caused the cluster to lose already "
"computed task(s), which will be recomputed elsewhere: {'x'} "
"(stimulus_id='ungraceful')",
Expand Down
Loading