Skip to content

remote/coordinator: prohibit multiple exporters with the same name #1775

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
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

remote/coordinator: prohibit multiple exporters with the same name #1775

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
29 changes: 29 additions & 0 deletions labgrid/remote/coordinator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -401,6 +401,8 @@ async def ExporterStream(self, request_iterator, context):
command_queue = asyncio.Queue()
pending_commands = []

startup_done = asyncio.Event()

out_msg = labgrid_coordinator_pb2.ExporterOutMessage()
out_msg.hello.version = labgrid_version()
yield out_msg
Expand All @@ -420,10 +422,15 @@ async def request_task():
elif kind == "startup":
version = in_msg.startup.version
name = in_msg.startup.name
if existing := self.get_exporter_by_name(name):
raise ExporterError(
f"exporter with name '{name}' is already connected from {existing.peer}"
)
session = self.exporters[peer] = ExporterSession(self, peer, name, command_queue, version)
logging.debug("Exporters: %s", self.exporters)
logging.debug("Received startup from %s with %s", name, version)
asyncio.current_task().set_name(f"exporter-{peer}-rx/started-{name}")
startup_done.set()
elif kind == "resource":
logging.debug("Received resource from %s with %s", name, in_msg.resource)
action, _ = session.set_resource(
Expand All @@ -439,10 +446,32 @@ async def request_task():
logging.debug("exporter request_task done: %s", context.done())
except Exception:
logging.exception("error in exporter message handler")
raise

asyncio.current_task().set_name(f"exporter-{peer}-tx")
running_request_task = self.loop.create_task(request_task(), name=f"exporter-{peer}-rx/init")

startup_done_task = self.loop.create_task(startup_done.wait())
done, _ = await asyncio.wait(
{startup_done_task, running_request_task},
timeout=3,
return_when=asyncio.FIRST_COMPLETED,
)
# clean up event task
startup_done.set()
await startup_done_task
if running_request_task in done:
# we probably had an exception during startup
try:
await running_request_task
except ExporterError as e:
await context.abort(grpc.StatusCode.ALREADY_EXISTS, f"startup failed: {e}")
raise
elif startup_done_task in done:
await startup_done_task
else:
raise ExporterError(f"exporter connection from {peer} timed out during startup")

try:
async for cmd in queue_as_aiter(command_queue):
logging.debug("exporter cmd %s", cmd)
Expand Down