Skip to content
This repository was archived by the owner on Jan 23, 2026. It is now read-only.

snmp: fix async handling #349

Merged
mangelajo merged 2 commits into from
Mar 16, 2025
Merged

snmp: fix async handling #349

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9ac28bc
snmp: fix async handling
bennyz Mar 15, 2025
0edf18c
Merge branch 'main' into snmp-fix
mangelajo Mar 16, 2025
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
50 changes: 34 additions & 16 deletions packages/jumpstarter-driver-snmp/jumpstarter_driver_snmp/driver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import socket
from dataclasses import dataclass, field
from enum import Enum, IntEnum
from typing import Any, Dict, Tuple

from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.entity import config, engine
Expand Down Expand Up @@ -116,9 +117,7 @@ def _setup_snmp(self):
def client(cls) -> str:
return "jumpstarter_driver_snmp.client.SNMPServerClient"

def _snmp_set(self, state: PowerState):
result = {"success": False, "error": None}

def _create_snmp_callback(self, result: Dict[str, Any], response_received: asyncio.Event):
def callback(snmpEngine, sendRequestHandle, errorIndication, errorStatus, errorIndex, varBinds, cbCtx):
self.logger.debug(f"Callback {errorIndication} {errorStatus} {errorIndex} {varBinds}")
if errorIndication:
Expand All @@ -135,20 +134,35 @@ def callback(snmpEngine, sendRequestHandle, errorIndication, errorStatus, errorI
for oid, val in varBinds:
self.logger.debug(f"{oid.prettyPrint()} = {val.prettyPrint()}")
self.logger.debug(f"SNMP set result: {result}")
response_received.set()

return callback

def _setup_event_loop(self) -> Tuple[asyncio.AbstractEventLoop, bool]:
try:
self.logger.info(f"Sending power {state.name} command to {self.host}")
created_loop = False
loop = asyncio.get_running_loop()
return loop, False
except RuntimeError:
loop = asyncio.new_event_loop()
Comment on lines +145 to +146
Copy link
Copy Markdown
Collaborator

@NickCao NickCao Mar 15, 2025

Choose a reason for hiding this comment

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

When would this happen?

Copy link
Copy Markdown
Member Author

@bennyz bennyz Mar 15, 2025

Choose a reason for hiding this comment

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

each time as we are not in the context of an event loop

asyncio.set_event_loop(loop)
return loop, True

async def _run_snmp_dispatcher(self, snmp_engine: engine.SnmpEngine, response_received: asyncio.Event):
snmp_engine.open_dispatcher()
await response_received.wait()
snmp_engine.close_dispatcher()

try:
asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
created_loop = True
def _snmp_set(self, state: PowerState):
result = {"success": False, "error": None}
response_received = asyncio.Event()
loop = None
created_loop = False

try:
self.logger.info(f"Sending power {state.name} command to {self.host}")
loop, created_loop = self._setup_event_loop()
snmp_engine = self._setup_snmp()

callback = self._create_snmp_callback(result, response_received)
cmdgen.SetCommandGenerator().send_varbinds(
snmp_engine,
"my-target",
Expand All @@ -158,11 +172,15 @@ def callback(snmpEngine, sendRequestHandle, errorIndication, errorStatus, errorI
callback,
)

snmp_engine.open_dispatcher(self.timeout)
snmp_engine.close_dispatcher()
dispatcher_task = loop.create_task(self._run_snmp_dispatcher(snmp_engine, response_received))
Copy link
Copy Markdown
Member

@mangelajo mangelajo Mar 15, 2025

Choose a reason for hiding this comment

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

I wonder if this dispatcher should be started before we send?

Is it used to receive the response?

I know it was like this before, but I wonder if it works just because it sets up very quickly before the response arrives?

Haven't checked the library docs though ...

Copy link
Copy Markdown
Member Author

@bennyz bennyz Mar 15, 2025

Choose a reason for hiding this comment

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

    def open_dispatcher(self, timeout: float = 0):
        """
        Open the dispatcher used by SNMP engine.

        This method is called when SNMP engine is ready to process SNMP
        messages. It opens the dispatcher and starts processing incoming
        messages.
        """
        if self.transport_dispatcher:
            self.transport_dispatcher.run_dispatcher(timeout)

In lib's docs they always call it last
https://github.com/lextudio/pysnmp/blob/8ddc451816c85d7ef238eb9cfdd5a74c114fae28/examples/v3arch/asyncio/manager/cmdgen/v2c-set.py#L12

Copy link
Copy Markdown
Member

@mangelajo mangelajo Mar 16, 2025

Choose a reason for hiding this comment

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

ahh, the dispatcher sends the messages and collects answers, ok :)

try:
loop.run_until_complete(asyncio.wait_for(dispatcher_task, self.timeout))
except asyncio.TimeoutError:
self.logger.warning(f"SNMP operation timed out after {self.timeout} seconds")
result["error"] = "SNMP operation timed out"

if not result["success"]:
raise SNMPError(result["error"])
raise SNMPError(result["error"] or "Unknown SNMP error")

return f"Power {state.name} command sent successfully"

Expand All @@ -171,7 +189,7 @@ def callback(snmpEngine, sendRequestHandle, errorIndication, errorStatus, errorI
self.logger.error(error_msg)
raise SNMPError(error_msg) from e
finally:
if created_loop:
if created_loop and loop:
loop.close()

@export
Expand Down