Add sam-cop tool - #314
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces sam-cop, a Python-based monitoring tool for SAM nodes that tracks peer connectivity, service churn, node events, and control plane health, delivering alerts via Slack, Telegram, or stdout. It also implements Go-side event buffering, adds new MCP tools (poll_node_events and get_control_plane_info), and updates the Python client to use SSE-friendly timeouts. The review feedback highlights several critical improvement opportunities: delivering alerts concurrently using asyncio.gather to prevent sequential blockages, adding defensive checks for null results in JSON parsing, avoiding memory and file descriptor leaks by closing abandoned clients asynchronously rather than storing them globally, catching standard exceptions instead of BaseException to allow clean shutdowns, and zeroing out discarded elements in the Go event buffer before re-slicing to prevent memory leaks.
| async def reconnect(client, state: SamCopState) -> Tuple[SamClient, str, SamCopState]: | ||
| """Recovers from a lost MCP session; the node's seq counter restarts too, so reset cursor.""" | ||
| _abandoned.append(client) |
There was a problem hiding this comment.
Parking dead clients in a global _abandoned list prevents them from ever being garbage collected, leading to a memory and file descriptor leak over time. Instead of keeping them alive indefinitely, you can close them asynchronously in a background task and safely swallow any exceptions.
| async def reconnect(client, state: SamCopState) -> Tuple[SamClient, str, SamCopState]: | |
| """Recovers from a lost MCP session; the node's seq counter restarts too, so reset cursor.""" | |
| _abandoned.append(client) | |
| async def reconnect(client, state: SamCopState) -> Tuple[SamClient, str, SamCopState]: | |
| """Recovers from a lost MCP session; the node's seq counter restarts too, so reset cursor.""" | |
| async def close_client_safely(c): | |
| try: | |
| await c.close() | |
| except Exception: | |
| pass | |
| asyncio.create_task(close_client_safely(client)) |
There was a problem hiding this comment.
The library only lets the task that opened a connection close it, so any close() here — even in a background task — throws and closes nothing; the suggestion just hides that error.
Dropping the reference is worse: the GC's cleanup then cancels the main polling loop (a bug I faced and this fixed). Parking the client is deliberate; it happens once per reconnect (after 3 failed cycles), so growth is negligible.
However, now I exit the process when len(_abandoned) > threshold so that a supervisor (such as k8s) can restart it with a clean state
|
I added the |
…security events, surface tool errors
…, add channel registry Move the agent to development/examples/sam-cop and rename Cop symbols to SamCop. Split the delivery backends into channels.py, where Channel subclasses self-register and declare required_env so build_channels picks them up without a registry edit. Node event categories and types are now exported constants instead of string literals at the call sites.
…meouts Handle SIGINT/SIGTERM via a shutdown flag so session-loss CancelledError is distinguishable from a real shutdown. Park dead clients instead of closing them cross-task. Always keep the stdout channel on. Use create_mcp_http_client so SSE streams don't hit httpx's 5s read timeout.
84f9590 to
5efce99
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces sam-cop, a Python-based watchdog and alerting tool that monitors SAM node events, detects network partitions, service churn, and control plane changes, and delivers alerts via Slack, Telegram, or stdout. It also adds Go backend support for recording and polling node events, exposes new MCP tools (poll_node_events and get_control_plane_info), and updates the Python client to use SSE-friendly timeouts. The review feedback highlights several performance and resource management improvements: asynchronously closing dead clients in sam_cop.py to prevent file descriptor leaks, using copy in events.go to preserve slice capacity and avoid repeated allocations, executing control plane fetches concurrently in mcp_handlers.go to reduce latency, and reusing a single httpx.AsyncClient in channels.py to avoid connection overhead.
| if len(b.entries) > nodeEventBufferSize { | ||
| discarded := len(b.entries) - nodeEventBufferSize | ||
| // Zero the trimmed prefix so its strings don't linger until the next reallocation. | ||
| for i := range discarded { | ||
| b.entries[i] = NodeEvent{} | ||
| } | ||
| b.entries = b.entries[discarded:] | ||
| } |
There was a problem hiding this comment.
Slicing the slice via b.entries = b.entries[discarded:] causes the slice's capacity to decrease over time. Once the capacity is reduced to nodeEventBufferSize, every subsequent record call will trigger a new underlying array allocation and memory copy. Shifting the elements to the beginning of the slice using copy preserves the slice's capacity at index 0, completely eliminating future allocations once the buffer is full.
| if len(b.entries) > nodeEventBufferSize { | |
| discarded := len(b.entries) - nodeEventBufferSize | |
| // Zero the trimmed prefix so its strings don't linger until the next reallocation. | |
| for i := range discarded { | |
| b.entries[i] = NodeEvent{} | |
| } | |
| b.entries = b.entries[discarded:] | |
| } | |
| if len(b.entries) > nodeEventBufferSize { | |
| discarded := len(b.entries) - nodeEventBufferSize | |
| copy(b.entries, b.entries[discarded:]) | |
| for i := len(b.entries) - discarded; i < len(b.entries); i++ { | |
| b.entries[i] = NodeEvent{} | |
| } | |
| b.entries = b.entries[:nodeEventBufferSize] | |
| } |
| info, err := FetchControlPlaneInfo(ctx, controlPlaneURL) | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
| keys, err := FetchControlPlaneKeys(ctx, controlPlaneURL) | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } |
There was a problem hiding this comment.
The calls to FetchControlPlaneInfo and FetchControlPlaneKeys are independent and can be executed concurrently to reduce the latency of the get_control_plane_info tool call. Since "sync" is already imported in this file, we can use a sync.WaitGroup to fetch both concurrently.
var info *api.ControlPlaneInfoResponse
var keys *api.KeysResponse
var errInfo, errKeys error
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
info, errInfo = FetchControlPlaneInfo(ctx, controlPlaneURL)
}()
go func() {
defer wg.Done()
keys, errKeys = FetchControlPlaneKeys(ctx, controlPlaneURL)
}()
wg.Wait()
if errInfo != nil {
return nil, nil, errInfo
}
if errKeys != nil {
return nil, nil, errKeys
}| def __init__(self, webhook_url: str): | ||
| self.webhook_url = webhook_url | ||
|
|
||
| async def send(self, message: str) -> None: | ||
| async with httpx.AsyncClient() as http_client: | ||
| response = await http_client.post(self.webhook_url, json={"text": message}, timeout=10.0) | ||
| response.raise_for_status() |
There was a problem hiding this comment.
Creating a new httpx.AsyncClient on every single alert delivery is inefficient as it incurs connection establishment and TLS handshake overhead for each message. We should instantiate a single httpx.AsyncClient in the channel's __init__ method and reuse it across all send calls.
| def __init__(self, webhook_url: str): | |
| self.webhook_url = webhook_url | |
| async def send(self, message: str) -> None: | |
| async with httpx.AsyncClient() as http_client: | |
| response = await http_client.post(self.webhook_url, json={"text": message}, timeout=10.0) | |
| response.raise_for_status() | |
| def __init__(self, webhook_url: str): | |
| self.webhook_url = webhook_url | |
| self.http_client = httpx.AsyncClient() | |
| async def send(self, message: str) -> None: | |
| response = await self.http_client.post(self.webhook_url, json={"text": message}, timeout=10.0) | |
| response.raise_for_status() |
| def __init__(self, bot_token: str, chat_id: str): | ||
| self.bot_token = bot_token | ||
| self.chat_id = chat_id | ||
|
|
||
| async def send(self, message: str) -> None: | ||
| url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" | ||
| async with httpx.AsyncClient() as http_client: | ||
| response = await http_client.post(url, json={"chat_id": self.chat_id, "text": message}, timeout=10.0) | ||
| response.raise_for_status() |
There was a problem hiding this comment.
Creating a new httpx.AsyncClient on every single alert delivery is inefficient as it incurs connection establishment and TLS handshake overhead for each message. We should instantiate a single httpx.AsyncClient in the channel's __init__ method and reuse it across all send calls.
| def __init__(self, bot_token: str, chat_id: str): | |
| self.bot_token = bot_token | |
| self.chat_id = chat_id | |
| async def send(self, message: str) -> None: | |
| url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" | |
| async with httpx.AsyncClient() as http_client: | |
| response = await http_client.post(url, json={"chat_id": self.chat_id, "text": message}, timeout=10.0) | |
| response.raise_for_status() | |
| def __init__(self, bot_token: str, chat_id: str): | |
| self.bot_token = bot_token | |
| self.chat_id = chat_id | |
| self.http_client = httpx.AsyncClient() | |
| async def send(self, message: str) -> None: | |
| url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" | |
| response = await self.http_client.post(url, json={"chat_id": self.chat_id, "text": message}, timeout=10.0) | |
| response.raise_for_status() |
|
Thanks @kaisoz — there are two changes here I'd like to land immediately. But I want to push back on the node-side event machinery: our goal is to give local agents visibility into mesh health without growing the MCP surface, and the node already has the vehicles for this. Please split out if you want to merge fast:
Request changes on the rest:
Suggested replacement (~5 lines, existing promauto pattern, no new tools):
Then re-base sam-cop on existing surfaces: scrape |
2 participants
Adds sam-cop, a mesh watchdog under development/tools/sam-cop that polls its local node over MCP.
Fixes #239