Skip to content

Use dedicated loggers for ClientSession/Place #1950

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
Bastian-Krause wants to merge 3 commits into
base: master
Choose a base branch
from
+26 −18
Open

Use dedicated loggers for ClientSession/Place #1950

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
394e652
remote/common: add logger argument to queue_as_aiter()
Bastian-Krause Jul 30, 2026
42b760b
remote/client: use dedicated logger for ClientSession
Bastian-Krause Jul 28, 2026
cd26faa
remote/common: use dedicated logger for Place
Bastian-Krause Jul 30, 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
30 changes: 16 additions & 14 deletions labgrid/remote/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,8 @@ def __attrs_post_init__(self):
self.sync_id = itertools.count(start=1)
self.sync_events = {}

self.logger = logging.getLogger("ClientSession")

@Emantor Emantor Aug 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Should we add a session identifier into the longer? f"ClientSession.{address}"?

@Bastian-Krause Bastian-Krause Aug 3, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I can't think of a case with multiple ClientSessions in a single run, so I guess it's not required.


async def start(self):
"""Starts receiving resource and place updates from the coordinator."""
self.resources = {}
Expand Down Expand Up @@ -173,13 +175,13 @@ async def sync_with_coordinator(self):
event = self.sync_events[identifier] = asyncio.Event()
msg = labgrid_coordinator_pb2.ClientInMessage()
msg.sync.id = identifier
logging.debug("sending sync %s", identifier)
self.logger.debug("sending sync %s", identifier)
self.out_queue.put_nowait(msg)
await event.wait()
if self.stopping.is_set():
logging.debug("sync %s failed", identifier)
self.logger.debug("sync %s failed", identifier)
else:
logging.debug("received sync %s", identifier)
self.logger.debug("received sync %s", identifier)
return not self.stopping.is_set()

def cancel_pending_syncs(self):
Expand All @@ -188,7 +190,7 @@ def cancel_pending_syncs(self):
while True:
try:
identifier, event = self.sync_events.popitem()
logging.debug("cancelling %s %s", identifier, event)
self.logger.debug("cancelling %s %s", identifier, event)
event.set()
except KeyError:
break
Expand All @@ -197,11 +199,11 @@ async def message_pump(self):
"""Task for receiving resource and place updates."""
got_message = False
try:
self.stream_call = call = self.stub.ClientStream(queue_as_aiter(self.out_queue))
self.stream_call = call = self.stub.ClientStream(queue_as_aiter(self.out_queue, self.logger))
async for out_msg in call:
out_msg: labgrid_coordinator_pb2.ClientOutMessage
got_message = True
logging.debug("out_msg from coordinator: %s", out_msg)
self.logger.debug("out_msg from coordinator: %s", out_msg)
for update in out_msg.updates:
update_kind = update.WhichOneof("kind")
if update_kind == "resource":
Expand All @@ -224,20 +226,20 @@ async def message_pump(self):
place_name = update.del_place
await self.on_place_deleted(place_name)
else:
logging.warning("unknown update from coordinator! %s", update_kind)
self.logger.warning("unknown update from coordinator! %s", update_kind)
if out_msg.HasField("sync"):
event = self.sync_events.pop(out_msg.sync.id)
event.set()
except grpc.aio.AioRpcError as e:
if e.code() == grpc.StatusCode.UNAVAILABLE:
if got_message:
logging.error("coordinator became unavailable: %s", e.details())
self.logger.error("coordinator became unavailable: %s", e.details())
else:
logging.error("coordinator is unavailable: %s", e.details())
self.logger.error("coordinator is unavailable: %s", e.details())
else:
logging.exception("unexpected grpc error in coordinator message pump task")
self.logger.exception("unexpected grpc error in coordinator message pump task")
except Exception:
logging.exception("error in coordinator message pump task")
self.logger.exception("error in coordinator message pump task")
finally:
self.stopping.set()
self.out_queue.put_nowait(None) # let the sender side exit gracefully
Expand Down Expand Up @@ -1065,13 +1067,13 @@ async def _console(self, place, target, timeout, *, logfile=None, loop=False, li
else:
call = ["telnet", host, str(port)]

logging.info("microcom not available, using telnet instead")
self.logger.info("microcom not available, using telnet instead")

if listen_only:
logging.warning("--listenonly option not supported by telnet, ignoring")
self.logger.warning("--listenonly option not supported by telnet, ignoring")

if logfile:
logging.warning("--logfile option not supported by telnet, ignoring")
self.logger.warning("--logfile option not supported by telnet, ignoring")

print(f"connecting to {resource} calling {' '.join(call)}")
try:
Expand Down
14 changes: 10 additions & 4 deletions labgrid/remote/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,9 @@ class Place:
changed = attr.ib(default=attr.Factory(time.time))
reservation = attr.ib(default=None)

def __attrs_post_init__(self):
self.logger = logging.getLogger(f"{self}")

def asdict(self):
# in the coordinator, we have resource objects, otherwise just a path
acquired_resources = []
Expand Down Expand Up @@ -354,7 +357,7 @@ def as_pb2(self):
place.tags[key] = value
return place
except TypeError:
logging.exception("failed to convert place %s to protobuf", self)
self.logger.exception("failed to convert place %s to protobuf", self)
raise

@classmethod
Expand All @@ -379,6 +382,9 @@ def from_pb2(cls, pb2):
reservation=pb2.reservation if pb2.HasField("reservation") else None,
)

def __str__(self):
return f"Place({self.name})"


class ReservationState(enum.Enum):
waiting = 0
Expand Down Expand Up @@ -481,7 +487,7 @@ def from_pb2(cls, pb2: labgrid_coordinator_pb2.Reservation):
)


async def queue_as_aiter(q):
async def queue_as_aiter(q, logger=logging.getLogger()):
try:
while True:
try:
Expand All @@ -493,7 +499,7 @@ async def queue_as_aiter(q):
return
yield item
q.task_done()
logging.debug("sent message %s", item)
logger.debug("sent message %s", item)
except Exception:
logging.exception("error in queue_as_aiter")
logger.exception("error in queue_as_aiter")
raise
Loading